UploadKit

FastAPI

Thin FastAPI adapters over Core. Choose async streaming (AsyncS3Storage / aioboto3) or sync (Boto3S3Storage / boto3). Pair with uploadkit-security.

Install

Shell 
pip install uploadkit-fastapi uploadkit-security
Shell 
uv add uploadkit-fastapi uploadkit-security
Shell 
poetry add uploadkit-fastapi uploadkit-security

Adapter glue

Shared policy, validators, errors, and JSON shape: Common patterns.

Copy AsyncS3Storage from Storage. Requires pip install aioboto3. Uses as_async_source() and background_after_upload().

main.py 
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()
# 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",
)

@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),
        )
    except UploaderError as exc:
        return json_error_response(exc)
    return {
        "object_name": result.object_name,
        "sha256": result.sha256,
        "etag": result.etag,
    }

Boto3S3Storage + run_sync_upload. Requires pip install boto3. Full class: Storage.

main.py 
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()
# AWS: Boto3S3Storage(access_key="AKIA...", secret_key="...", region="eu-west-1")
boto3_storage = Boto3S3Storage(
    endpoint_url="http://127.0.0.1:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
)

@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,
        "etag": result.etag,
    }

After-upload options

Pass Core after_upload on Uploader.upload / AsyncUploader.upload. The hook runs once after a successful put.

  1. BackgroundTasks — background_after_upload(background_tasks, notify) schedules work after the response is sent.
  2. Celery-like — object with .delay(**kwargs); Core calls delay(**result.as_task_kwargs()).
  3. Plain callback — sync (result) -> None, or async def on the async stack (awaited). Exceptions propagate and fail the request.

Shared semantics: Common patterns · Full storage classes: Storage · uploadkit-fastapi