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.
- ⏸️ 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
Install the latest release from PyPI:
# with uv
uv add tuspyserver
# with poetry
poetry add tuspyserver
# with pip
pip install tuspyserverOr install directly from source:
git clone https://github.com/edihasaj/tuspyserver
cd tuspyserver
pip install .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)
)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 metadataupload_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,
)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.
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,
)
)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,
)
)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,
)
)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()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.
To run the example, you need to install uv and run the following in the example/backend folder:
uv run server.pyThen, in another terminal window, run the following in example/frontend:
npm run devThis 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.
Contributions welcome! Please open issues or PRs on GitHub.
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
To release the package, follow the following steps:
- Update the version in
pyproject.tomlusing semver - Merge PR to main or push directly to main
- Open a PR to merge
main→production. - Upon merge, CI/CD will publish to PyPI.
© 2025 Edi Hasaj X