uploadkit-fastapi

FastAPI integration for UploadKit.

What problem does this solve?

Adapts Starlette/UploadFile for sync and async UploadKit stacks, bridges BackgroundTasks into Core after_upload, and maps exceptions to JSON responses — without reimplementing validation or storage.

When to use it

Use when your FastAPI app uploads files through UploadKit Core (Uploader or AsyncUploader).

When not to use it

Do not put validators, policies, or storage implementations in this package. Supply your own sync or async storage (boto3 / aioboto3 → AWS S3 or MinIO).

Choose sync or async

Installation

Requires Python 3.10+ and FastAPI 0.110+.

pip install uploadkit-fastapi uploadkit-security

uv add uploadkit-fastapi uploadkit-security

poetry add uploadkit-fastapi uploadkit-security

Storage samples (not package deps):

pip install boto3      # sync
pip install aioboto3   # async

Storage providers (AWS S3 and MinIO)

Copy the canonical Boto3S3Storage / AsyncS3Storage implementations from the uploadkit Core README (or the snippets below). AWS: omit endpoint_url. MinIO: set endpoint_url.

Sync — Boto3S3Storage

import boto3
from botocore.client import Config

class Boto3S3Storage:
    def __init__(self, *, access_key, secret_key, region="us-east-1", endpoint_url=None):
        kwargs = dict(
            service_name="s3",
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
            region_name=region,
            config=Config(signature_version="s3v4"),
        )
        if endpoint_url:
            kwargs["endpoint_url"] = endpoint_url
        self.client = boto3.client(**kwargs)

    def put(self, *, bucket, object_name, body, content_type):
        resp = self.client.put_object(
            Bucket=bucket, Key=object_name, Body=body, ContentType=content_type
        )
        return resp.get("ETag")

# AWS
boto3_storage = Boto3S3Storage(access_key="AKIA...", secret_key="...", region="eu-west-1")

# MinIO
boto3_storage = Boto3S3Storage(
    endpoint_url="http://127.0.0.1:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
)

Async — AsyncS3Storage (multipart)

Use the full AsyncS3Storage + AsyncS3Writer from the Core README (5 MiB part buffering). Construction:

# AWS
async_storage = AsyncS3Storage(access_key="AKIA...", secret_key="...", region="eu-west-1")

# MinIO
async_storage = AsyncS3Storage(
    endpoint_url="http://127.0.0.1:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
)

Async streaming route

from fastapi import BackgroundTasks, FastAPI, UploadFile
from uploadkit import AsyncUploader, UploadPolicy, UploaderError
from uploadkit_fastapi import (
    as_async_source,
    background_after_upload,
    json_error_response,
)
from uploadkit_security import default_async_validators

app = FastAPI()
# async_storage = AsyncS3Storage(...)  # AWS or MinIO — see above

def notify(result):
    ...

@app.post("/upload")
async def upload(file: UploadFile, background_tasks: BackgroundTasks):
    policy = UploadPolicy(
        max_size=5 * 1024 * 1024,
        allowed_extensions=frozenset({"png"}),
        allowed_mime_types=frozenset({"image/png"}),
        async_validators=default_async_validators(),
    )
    try:
        result = await AsyncUploader(policy, async_storage).upload(
            as_async_source(file),
            bucket="uploads",
            object_name=file.filename or "object",
            after_upload=background_after_upload(background_tasks, notify),
            # or after_upload=my_celery_task
            # or after_upload=sync_notify
        )
    except UploaderError as exc:
        return json_error_response(exc)
    return {
        "object_name": result.object_name,
        "sha256": result.sha256,
        "etag": result.etag,
    }

Sync stack (boto3) inside an async route

from fastapi import BackgroundTasks, FastAPI, UploadFile
from uploadkit import Uploader, UploadPolicy, UploaderError
from uploadkit_fastapi import (
    as_uploadable,
    background_after_upload,
    json_error_response,
    run_sync_upload,
)
from uploadkit_security import default_validators

app = FastAPI()
# boto3_storage = Boto3S3Storage(...)  # AWS or MinIO

@app.post("/upload-sync")
async def upload_sync(file: UploadFile, background_tasks: BackgroundTasks):
    policy = UploadPolicy(
        max_size=5 * 1024 * 1024,
        allowed_extensions=frozenset({"png"}),
        allowed_mime_types=frozenset({"image/png"}),
        validators=default_validators(),
    )
    try:
        result = await run_sync_upload(
            Uploader(policy, boto3_storage),
            as_uploadable(file),
            bucket="uploads",
            object_name=file.filename or "object",
            after_upload=background_after_upload(background_tasks, notify),
        )
    except UploaderError as exc:
        return json_error_response(exc)
    return {"object_name": result.object_name, "sha256": result.sha256}

run_sync_upload runs Uploader.upload in a worker thread.

Sync def route

@app.post("/upload-sync-def")
def upload_sync_def(file: UploadFile, background_tasks: BackgroundTasks):
    policy = UploadPolicy(
        max_size=5 * 1024 * 1024,
        validators=default_validators(),
    )
    try:
        result = Uploader(policy, boto3_storage).upload(
            as_uploadable(file),
            bucket="uploads",
            object_name=file.filename or "object",
            after_upload=background_after_upload(background_tasks, notify),
        )
    except UploaderError as exc:
        return json_error_response(exc)
    return {"object_name": result.object_name}

After-upload options

Pass Core after_upload on Uploader.upload / AsyncUploader.upload. The hook runs once after a successful put; it does not run on validation/storage failure.

1. BackgroundTasks — background_after_upload(background_tasks, notify) schedules notify after the response is sent. Prefer this for fire-and-forget work in FastAPI routes.
2. Celery-like — object with .delay(**kwargs); Core calls delay(**result.as_task_kwargs()) before returning.
3. Plain callback — sync (result) -> None, or async def on the async stack (awaited). Exceptions from a plain callback propagate and fail the request.

# Celery-like (sync or async stack)
result = await AsyncUploader(policy, async_storage).upload(
    as_async_source(file),
    bucket="uploads",
    object_name=file.filename or "object",
    after_upload=process_upload,  # process_upload.delay(**as_task_kwargs())
)

Full Core semantics: uploadkit Core README.

Public API

Changelog

See CHANGELOG.md.

Contributing

See CONTRIBUTING.md.