Core

Framework-free orchestration. Choose sync Uploader or async streaming AsyncUploader — then plug any framework adapter.

Install

Shell

pip install uploadkit uploadkit-security
# optional storage SDKs (not package deps):
# pip install boto3 aioboto3

Shell

uv add uploadkit uploadkit-security

Shell

poetry add uploadkit uploadkit-security

Examples

Shared UploadPolicy, validators, after-upload hooks, and error handling are documented in Common patterns.

Boto3S3Storage works for AWS S3 and MinIO. Full classes and install notes: Storage.

storage_sync.py

import boto3
from botocore.client import Config
from uploadkit import Uploader, UploadPolicy
from uploadkit_security import default_validators

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 S3
storage = Boto3S3Storage(access_key="AKIA...", secret_key="...", region="eu-west-1")

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

# Policy + validators: see Common patterns
policy = UploadPolicy(
    max_size=10 * 1024 * 1024,
    allowed_extensions=frozenset({"png", "jpg"}),
    allowed_mime_types=frozenset({"image/png", "image/jpeg"}),
    validators=default_validators(),
)
def notify(result):
    ...

result = Uploader(policy, storage).upload(
    file,
    bucket="uploads",
    object_name="2026/file.png",
    after_upload=notify,  # or Celery-like .delay
)

AsyncS3Storage (aioboto3 multipart) — same AWS vs MinIO wiring. Full writer class: Storage. Tune chunk_size and multipart part size in Performance.

example_async.py

from uploadkit import AsyncUploader, UploadPolicy
from uploadkit_security import default_async_validators
# from myapp.s3_async import AsyncS3Storage  # see /docs/storage/

# AWS S3
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",
)

policy = UploadPolicy(
    max_size=10 * 1024 * 1024,
    allowed_extensions=frozenset({"png", "jpg"}),
    allowed_mime_types=frozenset({"image/png", "image/jpeg"}),
    async_validators=default_async_validators(),
)

async def notify(result):
    ...

result = await AsyncUploader(policy, async_storage).upload(
    source,  # AsyncByteSource
    bucket="uploads",
    object_name="2026/file.png",
    after_upload=notify,  # sync, async, or Celery-like .delay
)

After-upload hooks

Optional after_upload on Uploader.upload / AsyncUploader.upload runs once after a successful store, before returning UploadResult. It does not run on validation or storage failure. Hook exceptions propagate.

Shapes: sync callback, async callback (AsyncUploader awaits), or Celery-like .delay(**result.as_task_kwargs()). Full shared write-up: Common patterns.

after_upload.py

def notify(result):
    ...

# Sync callback
Uploader(policy, storage).upload(
    file, bucket="uploads", object_name="a.png", after_upload=notify,
)

# Celery-like — Core calls task.delay(**result.as_task_kwargs())
Uploader(policy, storage).upload(
    file, bucket="uploads", object_name="a.png", after_upload=process_upload,
)

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

Storage providers: Storage · Tuning: Performance · Docs: uploadkit · uploadkit-security