Common patterns

Shared pieces used across Core and every framework guide. Framework pages only show adapter glue; put these conventions here once.

UploadPolicy

Size, extension, and MIME allow-lists live on the policy. Attach sync or async validators from uploadkit-security.

policy.py

from uploadkit import UploadPolicy
from uploadkit_security import default_validators, default_async_validators

# Sync pipeline
policy = UploadPolicy(
    max_size=5 * 1024 * 1024,
    allowed_extensions=frozenset({"png", "jpg"}),
    allowed_mime_types=frozenset({"image/png", "image/jpeg"}),
    validators=default_validators(),
)

# Async pipeline
async_policy = UploadPolicy(
    max_size=5 * 1024 * 1024,
    allowed_extensions=frozenset({"png"}),
    allowed_mime_types=frozenset({"image/png"}),
    async_validators=default_async_validators(),
)

After-upload hooks

Pass optional after_upload to Uploader.upload / AsyncUploader.upload. The hook runs once after a successful store, before returning UploadResult. It does not run on validation or storage failure. Hook exceptions propagate (not swallowed).

Accepted shapes:

Sync callback — (result: UploadResult) -> None
Async callback — sync or async def on AsyncUploader (awaited)
Celery-like — object with .delay(**kwargs); Core calls delay(**result.as_task_kwargs()) (no Celery import)
FastAPI — background_after_upload(background_tasks, notify) from uploadkit-fastapi

after_upload.py

def notify(result):
    # result.bucket, object_name, original_name, mime_type,
    # extension, size, sha256, etag
    ...

# Sync Uploader — callback or Celery-like .delay
Uploader(policy, storage).upload(
    file,
    bucket="uploads",
    object_name="a.png",
    after_upload=notify,  # or process_upload (has .delay)
)

# AsyncUploader — sync/async callback (awaited) or .delay
await AsyncUploader(policy, async_storage).upload(
    source,
    bucket="uploads",
    object_name="a.png",
    after_upload=notify,
)

# FastAPI — schedule via BackgroundTasks after the response
from uploadkit_fastapi import background_after_upload
after_upload=background_after_upload(background_tasks, notify)

# Celery kwargs mirror UploadResult.as_task_kwargs():
# bucket, object_name, original_name, mime_type, extension, size, sha256, etag

Error handling

Catch UploaderError. Django, FastAPI, Flask, and Odoo ship json_error_response; aiohttp (and custom stacks) map to the same JSON shape manually.

errors.py

from uploadkit import UploaderError

# Django / FastAPI / Flask / Odoo helpers
from uploadkit_django import json_error_response  # or uploadkit_fastapi / uploadkit_flask / uploadkit_odoo
try:
    result = Uploader(policy, storage).upload(...)
except UploaderError as exc:
    return json_error_response(exc)

# Manual JSON (e.g. aiohttp)
except UploaderError as exc:
    return web.json_response(
        {"error": type(exc).__name__, "message": str(exc)},
        status=400,
    )

Success JSON shape

Typical success payload returned by the sample views:

response.json

{
  "object_name": "2026/file.png",
  "sha256": "…",
  "etag": "…"
}

Validators

Use default_validators() / default_async_validators() from uploadkit-security. For MIME detection with libmagic, see the Security page.