Skip to content

UploadKit

Modular, framework-independent secure upload pipelines for Python.

Validate once. Upload anywhere. Same policy across Django, FastAPI, Flask, aiohttp, and Odoo.

Documentation · Website · PyPI


Package status

uploadkit (core)

CI Coverage Python

uploadkit-django

CI Coverage Python Django

uploadkit-fastapi

CI Coverage Python FastAPI

uploadkit-flask

CI Coverage Python Flask

uploadkit-odoo

CI Coverage Python Odoo

uploadkit-pdf

CI Coverage Python

uploadkit-audio

CI Coverage Python

uploadkit-office

CI Coverage Python

uploadkit-cli

CI Python


How the interface works

UploadKit separates policy, validation, storage, and framework I/O into clear seams. Core never depends on Django, FastAPI, Flask, Odoo, or any storage SDK.

  Framework file ──► adapter (as_uploadable / as_async_source)
                              │
                              ▼
                     UploadPolicy
                   (size, ext, MIME,
                    validators / async_validators)
                              │
              ┌───────────────┴───────────────┐
              ▼                               ▼
         Uploader                      AsyncUploader
              │                               │
              ▼                               ▼
    StorageProvider.put          AsyncStorageProvider.open_write
              │                               │
              └───────────────┬───────────────┘
                              ▼
                        UploadResult
                              │
                              ▼
                     after_upload hook
                    (callback / Celery .delay)

Core pieces

Piece Role
UploadPolicy Declares limits and which validators to run
Uploader / AsyncUploader Runs validation, then stores the object
UploadableFile / AsyncByteSource Protocols for the file/stream input
StorageProvider / AsyncStorageProvider Protocols you implement (e.g. boto3 / aioboto3 → S3 or MinIO)
UploadResult Bucket, object name, sha256, etag, … (as_task_kwargs() for queues)
after_upload Optional hook after a successful store
UploaderError Stable exception hierarchy; adapters map it to HTTP/JSON

Sync vs async

Sync:  Uploader.upload → validators → StorageProvider.put → UploadResult → after_upload
Async: AsyncUploader.upload → async validators → AsyncStorageProvider writer → UploadResult → after_upload

Quick Start

Requires Python 3.10+.

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

For the storage samples (not package dependencies):

pip install boto3          # sync AWS S3 / MinIO
pip install aioboto3       # async AWS S3 / MinIO

You supply storage / async_storage by implementing StorageProvider or AsyncStorageProvider (see the Core README for AWS S3 / MinIO samples).

Sync

from uploadkit import Uploader, UploadPolicy
from uploadkit_security import default_validators

policy = UploadPolicy(
    max_size=5 * 1024 * 1024,
    allowed_extensions=frozenset({"png"}),
    allowed_mime_types=frozenset({"image/png"}),
    validators=default_validators(),
)
result = Uploader(policy, storage).upload(
    file,  # UploadableFile
    bucket="uploads",
    object_name="2026/file.png",
)
# result.bucket, result.object_name, result.sha256, result.etag, …

Async streaming

from uploadkit import AsyncUploader, UploadPolicy
from uploadkit_security import default_async_validators

policy = UploadPolicy(
    max_size=5 * 1024 * 1024,
    allowed_extensions=frozenset({"png"}),
    allowed_mime_types=frozenset({"image/png"}),
    async_validators=default_async_validators(),
)
result = await AsyncUploader(policy, async_storage).upload(
    source,  # AsyncByteSource
    bucket="uploads",
    object_name="2026/file.png",
)

Optional after_upload on either path: a sync/async callback, or a Celery-like object with .delay(**result.as_task_kwargs()).


Integrate with your project (frameworks)

Install the adapter for your framework plus uploadkit-security. Each adapter only bridges framework files ↔ Core and maps UploaderError to responses — policies and storage stay yours.

Framework Package Docs
Django uploadkit-django Guide
FastAPI uploadkit-fastapi Guide
Flask uploadkit-flask Guide
Odoo uploadkit-odoo Guide
aiohttp Core AsyncByteSource directly Guide

Django

pip install uploadkit-django uploadkit-security
from django.conf import settings
from django.http import JsonResponse
from uploadkit import Uploader, UploadPolicy, UploaderError
from uploadkit_django import as_uploadable, get_storage_provider, json_error_response
from uploadkit_security import default_validators

def upload_view(request):
    storage = get_storage_provider()
    policy = UploadPolicy(
        max_size=5 * 1024 * 1024,
        allowed_extensions=frozenset({"png"}),
        allowed_mime_types=frozenset({"image/png"}),
        validators=default_validators(),
    )
    uploaded = request.FILES["file"]
    try:
        result = Uploader(policy, storage).upload(
            as_uploadable(uploaded),
            bucket=settings.UPLOADKIT_BUCKET,
            object_name=uploaded.name,
        )
    except UploaderError as exc:
        return json_error_response(exc)
    return JsonResponse({"object_name": result.object_name, "sha256": result.sha256})

Point UPLOADKIT_STORAGE_PROVIDER at your StorageProvider factory (boto3 for AWS/MinIO).

FastAPI

pip install uploadkit-fastapi uploadkit-security
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()

@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}

Use Uploader + as_uploadable + run_sync_upload when you prefer the sync stack inside an async route.

Flask

pip install uploadkit-flask uploadkit-security
from flask import current_app, jsonify, request
from uploadkit import Uploader, UploadPolicy, UploaderError
from uploadkit_flask import as_uploadable, get_storage_provider, json_error_response
from uploadkit_security import default_validators

@app.post("/upload")
def upload_view():
    storage = get_storage_provider()
    policy = UploadPolicy(
        max_size=5 * 1024 * 1024,
        allowed_extensions=frozenset({"png"}),
        allowed_mime_types=frozenset({"image/png"}),
        validators=default_validators(),
    )
    uploaded = request.files["file"]
    try:
        result = Uploader(policy, storage).upload(
            as_uploadable(uploaded),
            bucket=current_app.config["UPLOADKIT_BUCKET"],
            object_name=uploaded.filename,
        )
    except UploaderError as exc:
        return json_error_response(exc)
    return jsonify({"object_name": result.object_name, "sha256": result.sha256})

Odoo

pip install uploadkit-odoo uploadkit-security
from odoo import http
from odoo.http import request
from uploadkit import Uploader, UploadPolicy, UploaderError
from uploadkit_odoo import as_uploadable, json_error_response
from uploadkit_security import default_validators

class MyController(http.Controller):
    @http.route("/my/upload", type="http", auth="user", methods=["POST"], csrf=True)
    def upload(self, **kw):
        policy = UploadPolicy(
            max_size=5 * 1024 * 1024,
            allowed_extensions=frozenset({"png"}),
            allowed_mime_types=frozenset({"image/png"}),
            validators=default_validators(),
        )
        uploaded = kw.get("file")
        try:
            result = Uploader(policy, storage).upload(
                as_uploadable(uploaded),
                bucket="uploads",
                object_name=uploaded.filename,
            )
        except UploaderError as exc:
            return json_error_response(exc)
        return request.make_json_response(result.as_task_kwargs())

An optional Odoo 17/18 addon is included in the repo for settings and a thin upload service.


Ecosystem packages

Package Role
uploadkit Core pipeline
uploadkit-security MIME, filename, checksum validators
uploadkit-pdf / -office / -audio Feature policies
uploadkit-cli Developer CLI
uploadkit-testing Shared test utilities
uploadkit-specs RFCs & ADRs
uploadkit.github.io Docs site

Links

Pinned Loading

  1. uploadkit-django uploadkit-django Public

    Python

  2. uploadkit uploadkit Public

    Python

  3. uploadkit-fastapi uploadkit-fastapi Public

    Python

  4. uploadkit-odoo uploadkit-odoo Public

    Odoo integration for UploadKit

    Python

Repositories

Showing 10 of 19 repositories

People

This organization has no public members. You must be a member to see who’s a part of this organization.

Top languages

Loading…

Most used topics

Loading…