TuspyServer
TuspyServer is a Python library implementing the tus.io resumable upload protocol as a FastAPI router. It provides flexible backend storage options, including local disk (DiskStorage) and AWS S3 (S3Storage). The current version is 4.2.3, and it's actively maintained with regular updates and feature enhancements.
Warnings
- breaking Major refactor in v4.0.0 removed old storage implementations (e.g., `FileStorage`, `MemoryStorage`). Users must migrate to the new `DiskStorage` or `S3Storage` classes.
- breaking Python 3.7 support was dropped in v3.0.0. The library now requires Python 3.8 or newer for compatibility.
- gotcha When using `S3Storage`, the `boto3` library is a conditional dependency and must be installed explicitly using `pip install tuspyserver[s3]`.
- gotcha The `tuspyserver` integrates as a FastAPI router, meaning your application needs to be run using an ASGI server like Uvicorn. While `uvicorn` is a dependency, you still need to know how to execute it.
Install
-
pip install tuspyserver -
pip install tuspyserver[s3]
Imports
- TuspyServer
from tuspyserver import TuspyServer
- DiskStorage
from tuspyserver.storage import FileStorage
from tuspyserver import DiskStorage
- S3Storage
from tuspyserver import S3Storage
Quickstart
import os
from fastapi import FastAPI
from tuspyserver import TuspyServer, DiskStorage
app = FastAPI()
# Ensure the upload directory exists for DiskStorage
upload_dir = os.path.join(os.getcwd(), "upload_data")
os.makedirs(upload_dir, exist_ok=True)
# Initialize DiskStorage. For S3Storage, use:
# S3Storage(
# bucket_name=os.environ.get('S3_BUCKET_NAME', 'your-s3-bucket'),
# aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', ''),
# aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', ''),
# # region_name=os.environ.get('AWS_REGION', 'us-east-1')
# )
storage = DiskStorage(upload_dir)
# Initialize TuspyServer with the chosen storage backend
tus_server = TuspyServer(storage)
# Add the TuspyServer router to your FastAPI app
app.include_router(tus_server.router, prefix="/files")
# Optional: Add an endpoint to list uploaded files (for demonstration)
@app.get("/uploaded-files")
async def list_files():
return storage.list_files()
# To run this app, save it as main.py and execute:
# uvicorn main:app --reload --port 8000