Stop reimplementing file uploads for your Python ASGI app. Meet tussi.

Every few months I find myself in the same situation: "we need to support large file uploads". And every time, it's a chore.

This time, I decided to fix it properly to never have that headache again. I'm sharing my painkiller.

What I found

I discovered TUS. It's an open protocol for resumable file uploads. Clients pick up exactly where they left off after a network drop. There are TUS clients for multiple platforms and languages.

I checked out the existing Python server implementations. None of them fit my use case:

Most require a database to track upload state
Some need a separate daemon process
Multi-worker safety is either missing or poorly documented

So I decided, we need a better one.

Meet tussi

tussi is a TUS 1.0.0 resumable upload server for Python. ASGI-native, filesystem storage, no framework lock-in.

My USP: no database required, no separate daemon, and total thread/process/worker safety.

It works with any ASGI server, not just FastAPI. Drop it in, point a TUS client at it, done.

pip install tussi

Quickstart

from pathlib import Path
from tussi import TUSApp, FilesystemStorage

tus = TUSApp(
    storage=FilesystemStorage(directory=Path('./uploads')),
    completed_dir=Path('./completed'),
)

# tus is a standard ASGI callable
# run with: uvicorn myapp:tus

That's it. Works with any TUS client supporting protocol version 1.0.0.

FastAPI integration

tussi doesn't require FastAPI, but integrates cleanly:

from pathlib import Path
from fastapi import FastAPI, Request
from starlette.responses import Response
from tussi import TUSApp, FilesystemStorage

tus = TUSApp(
    storage=FilesystemStorage(directory=Path('./uploads')),
    completed_dir=Path('./completed'),
)
app = FastAPI()

@app.api_route(
    '/files/{path:path}',
    methods=['HEAD', 'PATCH', 'POST', 'OPTIONS'],
    include_in_schema=False,
)
async def tus_handler(request: Request) -> Response:
    return await tus.get_response(request.scope, request.receive)

Processing completed uploads

wait_for_file is an async context manager that blocks until a completed upload is available, claims it with an exclusive lock, and cleans up on exit. Safe to call from multiple concurrent workers.

async with tus.wait_for_file(timeout=3600) as upload:
    filename = upload.meta.get('filename', upload.name)
    upload.save(Path('./dest') / filename)

How it works under the hood

tussi uses posix_fallocate to pre-allocate disk space when an upload is created. No surprises when the disk fills up halfway through. fcntl.flock keeps concurrent workers from stepping on each other.

This is why it's Linux-only.

Event hooks

from tussi import TUSApp, TUSEvent, UploadCompletedEvent

async def on_event(event: TUSEvent) -> None:
    if isinstance(event, UploadCompletedEvent):
        print(f'upload complete: {event.upload_info.upload_id}')

tus = TUSApp(..., on_event=on_event)

Available events: UploadCreatedEvent, UploadProgressEvent, UploadCompletedEvent, UploadFailedEvent.

Try it out

pip install 'tussi[cli]'
tussi-server   # interactive demo server
tussi-upload   # CLI uploader for testing

It's my first entirely OSS library. I'd love feedback, especially on the API surface and what you'd need to actually use this in production.

推荐订阅源

DEV Community