aiohttp

No dedicated integration package — use Core directly. Wrap aiohttp multipart parts as AsyncByteSource and call AsyncUploader. Pair with uploadkit-security.

Install

Shell

pip install uploadkit uploadkit-security aiohttp

Shell

uv add uploadkit uploadkit-security aiohttp

Shell

poetry add uploadkit uploadkit-security aiohttp

Adapter glue

Shared policy and error conventions: Common patterns. Here the unique piece is the multipart AsyncByteSource.

Thin AsyncByteSource over aiohttp’s multipart body part (BodyPartReader).

adapters.py

import aiohttp
from aiohttp.multipart import BodyPartReader

class AiohttpByteSource:
    """Adapt an aiohttp multipart part for AsyncUploader."""

    def __init__(self, part: BodyPartReader) -> None:
        self._part = part
        self.name = part.filename
        self.size = None  # often unknown until fully read
        self.content_type = part.headers.get(
            aiohttp.hdrs.CONTENT_TYPE,
            "application/octet-stream",
        )

    async def read(self, size: int = -1) -> bytes:
        if size < 0:
            return await self._part.read(decode=False)
        chunk = await self._part.read_chunk(size)
        return chunk or b""

Parse multipart, find the file field, stream through AsyncUploader.

handlers.py

from aiohttp import web
from uploadkit import AsyncUploader, UploadPolicy, UploaderError
from uploadkit_security import default_async_validators

from adapters import AiohttpByteSource

async def upload_handler(request: web.Request) -> web.Response:
    reader = await request.multipart()
    part = await reader.next()
    while part is not None:
        if part.name == "file" and part.filename:
            break
        part = await reader.next()
    if part is None or not part.filename:
        return web.json_response(
            {"error": "MissingFile", "message": "file field required"},
            status=400,
        )

    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, request.app["async_storage"]).upload(
            AiohttpByteSource(part),
            bucket="uploads",
            object_name=part.filename,
            after_upload=notify,  # sync callback, async callback, or Celery-like .delay
        )
    except UploaderError as exc:
        return web.json_response(
            {"error": type(exc).__name__, "message": str(exc)},
            status=400,
        )
    return web.json_response({
        "object_name": result.object_name,
        "sha256": result.sha256,
        "etag": result.etag,
    })

Use AsyncS3Storage from Storage for AWS S3 or MinIO.

app.py

from aiohttp import web
from handlers import upload_handler
# from myapp.s3_async import AsyncS3Storage  # see /docs/storage/

async def notify(result) -> None:
    # Or pass a sync def, or a Celery-like task with .delay
    ...

# 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 = web.Application()
app["async_storage"] = async_storage
app.router.add_post("/upload", upload_handler)

if __name__ == "__main__":
    web.run_app(app, host="0.0.0.0", port=8080)

Same Core async stack as FastAPI — only the AsyncByteSource adapter differs. After-upload: Common patterns · Storage: Storage · Docs: uploadkit