or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

client-management.mdexception-handling.mdindex.mdrequest-handling.mdstorage-resumability.mdupload-operations.md
tile.json

tessl/pypi-tuspy

A Python client for the tus resumable upload protocol enabling pause and resume of file uploads

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/tuspy@1.1.x

To install, run

npx @tessl/cli install tessl/pypi-tuspy@1.1.0

index.mddocs/

TusPy

A Python client implementation for the tus resumable upload protocol, enabling robust file uploads with pause and resume capabilities. TusPy provides both synchronous and asynchronous operations, configurable chunk sizes, authentication support, and built-in retry mechanisms for handling network interruptions.

Package Information

  • Package Name: tuspy
  • Language: Python
  • Installation: pip install tuspy
  • Dependencies: requests, aiohttp, tinydb

Core Imports

from tusclient import client
from tusclient.uploader import Uploader, AsyncUploader

For storage and fingerprinting:

from tusclient.storage.interface import Storage
from tusclient.storage.filestorage import FileStorage
from tusclient.fingerprint.interface import Fingerprint
from tusclient.fingerprint.fingerprint import Fingerprint as DefaultFingerprint

For exceptions:

from tusclient.exceptions import TusCommunicationError, TusUploadFailed

For advanced request handling:

from tusclient.request import TusRequest, AsyncTusRequest, BaseTusRequest, catch_requests_error

Basic Usage

from tusclient import client

# Create client with server URL and optional headers
my_client = client.TusClient(
    'http://tusd.tusdemo.net/files/',
    headers={'Authorization': 'Bearer token'}
)

# Upload from file path
uploader = my_client.uploader('/path/to/file.ext', chunk_size=1024*1024)
uploader.upload()

# Upload from file stream
with open('/path/to/file.ext', 'rb') as fs:
    uploader = my_client.uploader(file_stream=fs, chunk_size=1024*1024)
    uploader.upload()

# Async upload
async_uploader = my_client.async_uploader('/path/to/file.ext', chunk_size=1024*1024)
await async_uploader.upload()

Architecture

TusPy implements the tus resumable upload protocol with a layered architecture:

  • Client Layer: TusClient manages server communication and provides uploader factories
  • Uploader Layer: BaseUploader, Uploader, and AsyncUploader handle file upload logic
  • Storage Layer: Optional URL storage for resumability across sessions using pluggable backends
  • Fingerprint Layer: File identification for resume capability using MD5-based fingerprinting
  • Request Layer: HTTP request abstraction supporting both requests and aiohttp libraries
  • Exception Layer: Custom exceptions for protocol-specific error handling

This design enables reliable file uploads over unstable connections with comprehensive error handling and retry mechanisms.

Capabilities

Client Management

Core client functionality for connecting to tus servers, managing authentication headers, and creating uploader instances.

class TusClient:
    def __init__(self, url: str, headers: Optional[Dict[str, str]] = None, 
                 client_cert: Optional[Union[str, Tuple[str, str]]] = None): ...
    def set_headers(self, headers: Dict[str, str]): ...
    def uploader(self, *args, **kwargs) -> Uploader: ...
    def async_uploader(self, *args, **kwargs) -> AsyncUploader: ...

Client Management

File Upload Operations

Synchronous and asynchronous file upload capabilities with chunked transfer, progress tracking, and error recovery.

class Uploader(BaseUploader):
    def upload(self, stop_at: Optional[int] = None): ...
    def upload_chunk(): ...
    def create_url(): ...

class AsyncUploader(BaseUploader):
    async def upload(self, stop_at: Optional[int] = None): ...
    async def upload_chunk(): ...
    async def create_url(): ...

File Upload Operations

Storage and Resumability

URL storage interfaces and implementations for resuming uploads across sessions.

class Storage(abc.ABC):
    def get_item(self, key: str): ...
    def set_item(self, key: str, value: str): ...
    def remove_item(self, key: str): ...

class FileStorage(Storage):
    def __init__(self, fp: str): ...
    def close(): ...

Storage and Resumability

Request Handling

HTTP request abstraction layer for tus protocol operations with both synchronous and asynchronous implementations.

class BaseTusRequest:
    def __init__(self, uploader): ...
    def add_checksum(self, chunk: bytes): ...

class TusRequest(BaseTusRequest):
    def perform(): ...

class AsyncTusRequest(BaseTusRequest):
    async def perform(): ...

def catch_requests_error(func): ...

Request Handling

Exception Handling

Custom exceptions for tus protocol communication errors and upload failures.

class TusCommunicationError(Exception):
    def __init__(self, message: str, status_code: Optional[int] = None, 
                 response_content: Optional[str] = None): ...

class TusUploadFailed(TusCommunicationError): ...

Exception Handling

Common Types

from typing import Dict, Optional, Tuple, Union, IO, Callable, List
import hashlib
import sys

# Type aliases used throughout the API
Headers = Dict[str, str]
ClientCert = Union[str, Tuple[str, str]]
FileInput = Union[str, IO]

# Constants used throughout the library
DEFAULT_HEADERS = {"Tus-Resumable": "1.0.0"}
DEFAULT_CHUNK_SIZE = sys.maxsize
CHECKSUM_ALGORITHM_PAIR = ("sha1", hashlib.sha1)
FINGERPRINT_BLOCK_SIZE = 65536