tuspyserver

A FastAPI router implementing a tus upload protocol server, with optional dependency-injected hooks for post-upload processing.

Only depends on fastapi>=0.110 and python>=3.8.

Features

• ⏸️ Resumable uploads via TUS protocol
• 🍰 Chunked transfer with configurable max size
• 🗃️ Metadata storage (filename, filetype)
• 🧹 Expiration & cleanup of old uploads (default retention: 5 days)
• 💉 Dependency injection for seamless validation (optional)
• 📡 Comprehensive API with download, HEAD, DELETE, and OPTIONS endpoints
• 🗄️ Pluggable storage — local filesystem by default, or stream straight to S3

Installation

Install the latest release from PyPI:

# with uv
uv add tuspyserver
# with poetry
poetry add tuspyserver
# with pip
pip install tuspyserver

Or install directly from source:

git clone https://github.com/edihasaj/tuspyserver
cd tuspyserver
pip install .

For the S3 storage backend, install the s3 extra (it pulls in boto3):

pip install "tuspyserver[s3]"

Usage

API

The main API is a single constructor that initializes the tus router. All arguments are optional, and these are their default values:

from tuspyserver import create_tus_router

tus_router = create_tus_router(
    prefix="files",                                   # route prefix (default: 'files')
    files_dir="/tmp/files",                  # path to store files
    max_size=128_849_018_880,             # max upload size in bytes (default is ~128GB)
    auth=noop,                                              # authentication dependency
    days_to_keep=5,                                   # retention period
    on_upload_complete=None,               # upload callback
    upload_complete_dep=None,             # upload callback (dependency injector)
    pre_create_hook=None,                 # pre-creation callback
    pre_create_dep=None,                  # pre-creation callback (dependency injector)
    file_dep=None,                        # file path callback (dependency injector)
)

Pre-Create Hook

The Pre-Create Hook allows you to validate metadata and perform authentication before a file is created on the server. This is useful for:

• Metadata validation: Check if required fields are present, validate file types, etc.
• User authentication: Verify user permissions before allowing upload creation
• Business logic: Apply custom rules before file creation

The hook receives two parameters:

• metadata: A dictionary containing the decoded upload metadata
• upload_info: A dictionary with upload parameters (size, defer_length, expires)

def validate_upload(metadata: dict, upload_info: dict):
    # Validate required metadata
    if "filename" not in metadata:
        raise HTTPException(status_code=400, detail="Filename is required")
    
    # Check file size limits
    if upload_info["size"] and upload_info["size"] > 100_000_000:  # 100MB
        raise HTTPException(status_code=413, detail="File too large")
    
    # Validate file type
    if "filetype" in metadata:
        allowed_types = ["image/jpeg", "image/png", "application/pdf"]
        if metadata["filetype"] not in allowed_types:
            raise HTTPException(status_code=400, detail="File type not allowed")

# Use the hook
tus_router = create_tus_router(
    files_dir="./uploads",
    pre_create_hook=validate_upload,
)

Basic setup

In your main.py:

from tuspyserver import create_tus_router

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles

import uvicorn

# initialize a FastAPI app
app = FastAPI()

# configure cross-origin middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=[
        "Location",
        "Upload-Offset",
        "Tus-Resumable",
        "Tus-Version",
        "Tus-Extension",
        "Tus-Max-Size",
        "Upload-Expires",
        "Upload-Length",
    ],
)

# use completion hook to log uploads
def log_upload(file_path: str, metadata: dict):
    print("Upload complete")
    print(file_path)
    print(metadata)


# mount the tus router to our
app.include_router(
    create_tus_router(
        files_dir="./uploads",
        on_upload_complete=log_upload,
    )
)

[!IMPORTANT] Headers must be exposed for chunked uploads to work correctly.

For a comprehensive working example, see the tuspyserver example.

Dependency injection

For applications using FastAPI's dependency injection, you can supply a factory function that returns a callback with injected dependencies. The factory can Depends() on any of your services (database session, current user, etc.).

# Define a factory dependency that injects your own services
from fastapi import Depends
from your_app.dependencies import get_db, get_current_user

# factory function
def log_user_upload(
    db=Depends(get_db),
    current_user=Depends(get_current_user),
) -> Callable[[str, dict], None]:
    # callback function
    async def handler(file_path: str, metadata: dict):
        # perform validation or post-processing
        await db.log_upload(current_user.id, metadata)
        await process_file(file_path)
    return handler

# Include router with the DI hook
app.include_router(
    create_api_router(
        upload_complete_dep=log_user_upload,
    )
)

Pre-Create Hook with Dependency Injection

You can also use dependency injection with the Pre-Create Hook for authentication and validation:

from fastapi import Depends, HTTPException
from your_app.dependencies import get_db, get_current_user

def validate_user_upload(
    db=Depends(get_db),
    current_user=Depends(get_current_user),
) -> Callable[[dict, dict], None]:
    # callback function
    async def handler(metadata: dict, upload_info: dict):
        # Check user permissions
        if not current_user.can_upload:
            raise HTTPException(status_code=403, detail="Upload not allowed")
        
        # Validate against user's quota
        user_uploads = await db.get_user_uploads(current_user.id)
        if len(user_uploads) >= current_user.upload_limit:
            raise HTTPException(status_code=429, detail="Upload quota exceeded")
        
        # Log the upload attempt
        await db.log_upload_attempt(current_user.id, metadata, upload_info)
    
    return handler

# Include router with the pre-create DI hook
app.include_router(
    create_tus_router(
        pre_create_dep=validate_user_upload,
    )
)

File Routing Dependency Injection

You can use dependency injection with file dep for directly storing the file:

from fastapi import Depends, HTTPException
from your_app.dependencies import get_db, get_current_user, get_user_dir

def get_file(
    db=Depends(get_db),
    current_user=Depends(get_current_user),
) -> Callable[[dict, dict], None]:
    # callback function
    async def handler(metadata: dict):
        # Get the file name
        file_name = metadata["file_name"]
        # Get the file directory
        file_dir = get_user_dir(current_user)

        return {
            "file_dir": file_dir,
            "uid": file_name
        }

    return handler

# Include router with the pre-create DI hook
app.include_router(
    create_tus_router(
        file_dep=file_dep,
    )
)

Storage backends

By default uploads are written to files_dir on the local filesystem, exactly as they always have been. Passing storage= swaps in a different backend.

Why you might want one

Sharing files_dir across replicas usually means a network filesystem, and that couples every request to it. Two things follow:

• A slow or wedged mount blocks the event loop, because the filesystem calls in an async route are synchronous. A stalled worker stops answering its health endpoint, so an orchestrator marks the whole replica unready.
• The shared volume becomes a single point of failure, and an RWX volume is often the most fragile piece of a deployment.

Both storage backends here keep their I/O off the event loop, and the S3 backend removes the shared filesystem entirely.

S3

Uploads stream into an S3 multipart upload. No shared volume is involved, and any replica can serve any chunk of any upload.

import boto3
from tuspyserver import create_tus_router
from tuspyserver.storage.s3 import S3Storage

s3 = boto3.client(
    "s3",
    endpoint_url="https://s3.eu-west-par.io.cloud.ovh.net",  # any S3-compatible store
    region_name="eu-west-par",
    aws_access_key_id="...",
    aws_secret_access_key="...",
)

tus_router = create_tus_router(
    prefix="files",
    storage=S3Storage(bucket="my-bucket", client=s3, prefix="tus/"),
    on_upload_complete=lambda location, metadata: print(location, metadata),
)

You pass the client in, so credentials, endpoints and retry policy stay under your control and an existing client can be reused.

Chunk sizes. S3 requires every part except the last to be at least 5 MiB. tus lets a client PATCH any size it likes, so anything smaller is parked in a .part object and prepended to the next PATCH. That is correct but costs an extra round trip per chunk, so prefer a client chunk size of 5 MiB or more:

// tus-js-client / Uppy — note MiB, not MB: 5 * 1000 * 1000 is below the limit
new tus.Upload(file, { chunkSize: 5 * 1024 * 1024 })

part_size (default 5 MiB) sets the flush threshold. A multipart upload is capped at 10,000 parts, so the largest uploadable file is part_size * 10_000 — 50 GiB at the default.

on_upload_complete receives a location, not a path. With S3Storage it is s3://<bucket>/<key>, so a hook must not assume it can open() the value. The local backend still passes a filesystem path.

Locking. A PATCH is read-modify-write on the upload offset, so two requests for one upload must not interleave. S3Storage serializes them within a single process; across replicas S3 offers no lock primitive, leaving only the tus Upload-Offset precondition, which is itself read-then-write.

Run more than one replica and you want lock_factory — any uid -> async context manager:

import contextlib

@contextlib.asynccontextmanager
async def upload_lock(uid: str):
    async with my_redis_lock(f"tus:{uid}", ttl=300):
        yield

S3Storage(bucket="...", client=s3, lock_factory=upload_lock)

Both locks are taken when a factory is present; the in-process one saves a round trip for same-process contention. Give the lock a TTL comfortably longer than a single PATCH — a slow client sending a large chunk can hold it for minutes — so that a pod dying mid-request releases it.

Concatenation is not supported by S3Storage; a final concatenated upload returns 501. The other extensions — creation, creation-with-upload, expiration, termination — all work.

Local filesystem

The same on-disk layout as the default, but with the syscalls dispatched off the event loop. Useful if you keep a shared volume but want a slow mount to degrade uploads instead of the whole worker.

from tuspyserver.storage.local import LocalFileStorage

tus_router = create_tus_router(storage=LocalFileStorage("/data/uploads"))

Writing your own

Subclass tuspyserver.storage.TusStorage and implement its async methods — create, exists, size, append, flush, finalize, read, delete, read_info, write_info, list_uids, location and lock. Keep blocking work off the event loop (asyncio.to_thread is enough).

Expiration & cleanup

Expired files are removed when remove_expired_files() is called. You can schedule it using your preferred background scheduler (e.g., APScheduler, cron).

from tuspyserver import create_tus_router

from apscheduler.schedulers.background import BackgroundScheduler

tus_router = create_tus_router(
    days_to_keep = 23  # configure retention period; defaults to 5 days
)

scheduler = BackgroundScheduler()
scheduler.add_job(
    lambda: tus_router.remove_expired_files(),
    trigger='cron',
    hour=1,
)
scheduler.start()

Example

You can find a complete working basic example in the example folder.

the example consists of a backend serving fastapi with uvicorn, and a frontend npm project.

Running the example

To run the example, you need to install uv and run the following in the example/backend folder:

uv run server.py

Then, in another terminal window, run the following in example/frontend:

npm run dev

This should launch the server, and you should now be able to test uploads by browsing to http://localhost:5173.

Uploaded files get placed in the example/backend/uploads folder.

Developing

Contributions welcome! Please open issues or PRs on GitHub. The proposed contribution of tuspyserver to the tus organization is tracked in tus/tus.io#525.

You need uv to develop the project. The project is setup as a uv workspace where the root is the library and the example directory is an unpackaged app

Releasing

Releases are fully automated. Every push to main runs .github/workflows/publish.yml, which:

1. Computes the next version — takes the higher of the pyproject.toml version and the latest release on PyPI, then walks up to the first unused patch. PyPI is the source of truth for what has already shipped.
2. Sets that version, builds, and publishes to PyPI using the PYPI_API_TOKEN repository secret.

So to ship a patch release, just merge to main — no manual version bump needed. For a minor/major release, bump the version in pyproject.toml in the same change and CI will publish from there. The version in pyproject.toml acts as a floor and does not need to track every published patch.

© 2025 Edi Hasaj X