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

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 .

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,
    )
)

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