Metadata-Version: 2.5
Name: uploadkit-fastapi
Version: 0.1.1
Summary: FastAPI integration for UploadKit
Project-URL: Homepage, https://github.com/uploadkit/uploadkit-fastapi
Project-URL: Repository, https://github.com/uploadkit/uploadkit-fastapi
Author: UploadKit
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: fastapi,upload,uploadkit
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: fastapi>=0.110
Requires-Dist: uploadkit>=0.2.0
Provides-Extra: dev
Requires-Dist: coverage>=7.0; extra == 'dev'
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: uploadkit-security>=0.2.0; extra == 'dev'
Requires-Dist: uploadkit-testing>=0.1.0; extra == 'dev'
Provides-Extra: security
Requires-Dist: uploadkit-security>=0.2.0; extra == 'security'
Description-Content-Type: text/markdown

# uploadkit-fastapi

[![CI](https://github.com/uploadkit/uploadkit-fastapi/actions/workflows/ci.yml/badge.svg)](https://github.com/uploadkit/uploadkit-fastapi/actions/workflows/ci.yml)
[![Coverage](https://img.shields.io/badge/coverage-100%25-brightgreen)](https://github.com/uploadkit/uploadkit-fastapi/actions/workflows/ci.yml)
[![Python](https://img.shields.io/badge/python-3.10%20%7C%203.11%20%7C%203.12%20%7C%203.13-blue)](pyproject.toml)
[![FastAPI](https://img.shields.io/badge/fastapi-0.110%2B-teal)](pyproject.toml)

FastAPI integration for UploadKit.

## What problem does this solve?

Adapts Starlette/`UploadFile` for sync and async UploadKit stacks, bridges `BackgroundTasks` into Core `after_upload`, and maps exceptions to JSON responses — without reimplementing validation or storage.

## When to use it

Use when your FastAPI app uploads files through UploadKit Core (`Uploader` or `AsyncUploader`).

## When not to use it

Do not put validators, policies, or storage implementations in this package. Supply your own sync or async storage (boto3 / aioboto3 → AWS S3 or MinIO).

## Choose sync or async

| | **Async streaming** | **Sync** |
|--|---------------------|----------|
| Entry | `AsyncUploader` | `Uploader` |
| File adapter | `as_async_source(file)` | `as_uploadable(file)` |
| Validators | `async_validators=default_async_validators()` | `validators=default_validators()` |
| Storage | `AsyncS3Storage` (aioboto3) | `Boto3S3Storage` (boto3) |
| In an `async def` route | `await AsyncUploader(...).upload(...)` | `await run_sync_upload(...)` |

## Installation

Requires **Python 3.10+** and **FastAPI 0.110+**.

```bash
pip install uploadkit-fastapi uploadkit-security
```

```bash
uv add uploadkit-fastapi uploadkit-security
```

```bash
poetry add uploadkit-fastapi uploadkit-security
```

Storage samples (not package deps):

```bash
pip install boto3      # sync
pip install aioboto3   # async
```

## Storage providers (AWS S3 and MinIO)

Copy the canonical `Boto3S3Storage` / `AsyncS3Storage` implementations from the
[uploadkit Core README](https://github.com/uploadkit/uploadkit#storage-examples-aws-s3-and-minio)
(or the snippets below). **AWS:** omit `endpoint_url`. **MinIO:** set `endpoint_url`.

### Sync — `Boto3S3Storage`

```python
import boto3
from botocore.client import Config

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

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

### Async — `AsyncS3Storage` (multipart)

Use the full `AsyncS3Storage` + `AsyncS3Writer` from the
[Core README](https://github.com/uploadkit/uploadkit#storage-examples-aws-s3-and-minio)
(5 MiB part buffering). Construction:

```python
# 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",
)
```

## Async streaming route

```python
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()
# async_storage = AsyncS3Storage(...)  # AWS or MinIO — see above

def notify(result):
    ...

@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),
            # or after_upload=my_celery_task
            # or after_upload=sync_notify
        )
    except UploaderError as exc:
        return json_error_response(exc)
    return {
        "object_name": result.object_name,
        "sha256": result.sha256,
        "etag": result.etag,
    }
```

## Sync stack (boto3) inside an async route

```python
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()
# boto3_storage = Boto3S3Storage(...)  # AWS or MinIO

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

`run_sync_upload` runs `Uploader.upload` in a worker thread.

### Sync `def` route

```python
@app.post("/upload-sync-def")
def upload_sync_def(file: UploadFile, background_tasks: BackgroundTasks):
    policy = UploadPolicy(
        max_size=5 * 1024 * 1024,
        validators=default_validators(),
    )
    try:
        result = Uploader(policy, boto3_storage).upload(
            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}
```

## After-upload options

Pass Core `after_upload` on `Uploader.upload` / `AsyncUploader.upload`. The hook runs once after a successful put; it does not run on validation/storage failure.

1. **BackgroundTasks** — `background_after_upload(background_tasks, notify)` schedules `notify` after the response is sent. Prefer this for fire-and-forget work in FastAPI routes.
2. **Celery-like** — object with `.delay(**kwargs)`; Core calls `delay(**result.as_task_kwargs())` before returning.
3. **Plain callback** — sync `(result) -> None`, or `async def` on the async stack (awaited). Exceptions from a plain callback **propagate** and fail the request.

```python
# Celery-like (sync or async stack)
result = await AsyncUploader(policy, async_storage).upload(
    as_async_source(file),
    bucket="uploads",
    object_name=file.filename or "object",
    after_upload=process_upload,  # process_upload.delay(**as_task_kwargs())
)
```

Full Core semantics: [uploadkit Core README](https://github.com/uploadkit/uploadkit#after-upload-hooks).

## Public API

| Symbol | Kind |
|--------|------|
| `as_uploadable` | Sync `UploadFile` → `UploadableFile` |
| `as_async_source` | `UploadFile` → `AsyncByteSource` |
| `background_after_upload` | Core hook via FastAPI `BackgroundTasks` |
| `run_sync_upload` | `asyncio.to_thread` around sync `Uploader.upload` |
| `json_error_response` / `status_for_error` / `error_payload` | Public |

## Changelog

See [CHANGELOG.md](CHANGELOG.md).

## Contributing

See [CONTRIBUTING.md](CONTRIBUTING.md).
