Metadata-Version: 2.3
Name: tuspyserver
Version: 4.3.0
Summary: A Python tus server implementation as a FastAPI router
Author: Edi Hasaj
Author-email: Edi Hasaj <edi.hasaj@applifyer.com>
Requires-Dist: fastapi>=0.110.0
Requires-Dist: boto3>=1.26 ; extra == 's3'
Requires-Python: >=3.8
Provides-Extra: s3
Description-Content-Type: text/markdown

<a href="https://pypi.org/project/tuspyserver/"><img alt="PyPI - Version" src="https://img.shields.io/pypi/v/tuspyserver" align="right"></a>

# tuspyserver

A FastAPI router implementing a [tus upload protocol](https://tus.io/) 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](https://pypi.org/project/tuspyserver/):

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

Or install directly from source:

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

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

```bash
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:

```python
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)

```python
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`:

```python
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](#example).

### Dependency injection

For applications using FastAPI's [dependency injection](https://fastapi.tiangolo.com/tutorial/dependencies/), 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.).

```python
# 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:

```python
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:

```python
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.

```python
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:

```js
// 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.** `S3Storage` serializes concurrent PATCHes for one upload within a
single process. Across replicas it relies on the tus `Upload-Offset`
precondition, which rejects a racing PATCH with `409`. If you need a hard
cross-replica guarantee, wrap the backend with an external lock.

**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.

```python
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`).

```python
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](https://github.com/edihasaj/tuspyserver/tree/main/examples) 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`](https://docs.astral.sh/uv/) and run the following in the `example/backend` folder:
```bash
uv run server.py
```

Then, in another terminal window, run the following in `example/frontend`:
```bash
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](https://github.com/edihasaj/tuspyserver).
The proposed contribution of tuspyserver to the tus organization is tracked in
[tus/tus.io#525](https://github.com/tus/tus.io/issues/525).

You need [`uv`](https://docs.astral.sh/uv/) to develop the project. The project is setup as a [uv workspace](https://docs.astral.sh/uv/concepts/projects/workspaces/)
where the root is the [library](https://docs.astral.sh/uv/concepts/projects/init/#libraries) and the example directory is an [unpackaged app](https://docs.astral.sh/uv/concepts/projects/init/#applications)

### Releasing

Releases are fully automated. Every push to `main` runs
[`.github/workflows/publish.yml`](.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](https://x.com/hasajedi)*
