tessl/pypi-tuspy
A Python client for the tus resumable upload protocol enabling pause and resume of file uploads
upload-operations.mddocs/
File Upload Operations
Synchronous and asynchronous file upload capabilities with chunked transfer, progress tracking, and error recovery. TusPy provides both sync and async uploaders built on a common base class.
Capabilities
Base Uploader
Common functionality shared by synchronous and asynchronous uploaders.
class BaseUploader:
"""
Object to control upload related functions.
Attributes:
file_path (str): Path to file for upload
file_stream (IO): Alternative file stream for upload
url (str): Upload URL if known (for resuming)
client (TusClient): Client instance for server communication
chunk_size (int): Size of chunks in bytes (default: sys.maxsize)
metadata (dict): Upload metadata dictionary
metadata_encoding (str): Encoding for metadata values (default: 'utf-8')
offset (int): Current upload position
stop_at (int): Offset to stop upload at
retries (int): Number of retry attempts on failure (default: 0)
retry_delay (int): Delay between retries in seconds (default: 30)
verify_tls_cert (bool): Whether to verify TLS certificates (default: True)
store_url (bool): Whether to store URL for resumability (default: False)
url_storage (Storage): Storage implementation for URL persistence
fingerprinter (Fingerprint): Fingerprint implementation for file identification
upload_checksum (bool): Whether to include upload checksums (default: False)
"""
DEFAULT_HEADERS = {"Tus-Resumable": "1.0.0"}
DEFAULT_CHUNK_SIZE = sys.maxsize
CHECKSUM_ALGORITHM_PAIR = ("sha1", hashlib.sha1)
def __init__(self, file_path: Optional[str] = None, file_stream: Optional[IO] = None,
url: Optional[str] = None, client: Optional[TusClient] = None,
chunk_size: int = sys.maxsize, metadata: Optional[Dict] = None,
metadata_encoding: Optional[str] = "utf-8", retries: int = 0,
retry_delay: int = 30, verify_tls_cert: bool = True,
store_url: bool = False, url_storage: Optional[Storage] = None,
fingerprinter: Optional[Fingerprint] = None, upload_checksum: bool = False):
"""
Initialize BaseUploader.
Parameters:
- file_path (Optional[str]): Path to file for upload
- file_stream (Optional[IO]): File stream alternative to file_path
- url (Optional[str]): Known upload URL for resuming
- client (Optional[TusClient]): Client instance for server communication
- chunk_size (int): Chunk size in bytes for uploads
- metadata (Optional[Dict]): Upload metadata dictionary
- metadata_encoding (Optional[str]): Encoding for metadata values
- retries (int): Number of retry attempts on failure
- retry_delay (int): Delay between retries in seconds
- verify_tls_cert (bool): Whether to verify TLS certificates
- store_url (bool): Whether to store URL for resumability
- url_storage (Optional[Storage]): Storage for URL persistence
- fingerprinter (Optional[Fingerprint]): File fingerprinting implementation
- upload_checksum (bool): Whether to include upload checksums
"""Header Management
Methods for managing HTTP headers in uploads.
def get_headers(self) -> Dict[str, str]:
"""
Return headers of the uploader instance including client headers.
Returns:
Dict[str, str]: Combined headers from uploader and client
"""
def get_url_creation_headers(self) -> Dict[str, str]:
"""
Return headers required to create upload url.
Returns:
Dict[str, str]: Headers for POST request to create upload URL
"""File Operations
Methods for working with files and file streams.
def get_file_stream(self) -> IO:
"""
Return a file stream instance of the upload.
Returns:
IO: File stream positioned at start of file
"""
def get_file_size(self) -> int:
"""
Return size of the file in bytes.
Returns:
int: File size in bytes
"""
def get_request_length(self) -> int:
"""
Return length of next chunk upload in bytes.
Returns:
int: Length of next chunk to upload
"""URL and Offset Management
Methods for managing upload URLs and tracking progress.
def set_url(self, url: str):
"""
Set the upload URL and optionally store it.
Parameters:
- url (str): Upload URL from server
"""
def get_offset(self) -> int:
"""
Return current offset from tus server.
Makes HTTP HEAD request to retrieve current upload position.
Returns:
int: Current upload offset in bytes
"""Metadata Handling
Methods for encoding and managing upload metadata.
def encode_metadata(self) -> List[str]:
"""
Return list of encoded metadata as defined by the Tus protocol.
Returns:
List[str]: Encoded metadata strings for upload-metadata header
"""Properties
Read-only properties for accessing uploader state and configuration.
@property
def checksum_algorithm(self) -> Callable:
"""
Return checksum algorithm function for upload verification.
Returns:
Callable: Checksum algorithm function (default: hashlib.sha1)
"""
@property
def checksum_algorithm_name(self) -> str:
"""
Return name of checksum algorithm for upload verification.
Returns:
str: Algorithm name (default: "sha1")
"""
@property
def client_cert(self) -> Optional[Union[str, Tuple[str, str]]]:
"""
Return client certificate configuration from associated client.
Returns:
Optional[Union[str, Tuple[str, str]]]: Client certificate path(s) or None
"""Synchronous Uploader
Synchronous file uploader using the requests library.
class Uploader(BaseUploader):
"""Synchronous file uploader using requests library."""
def upload(self, stop_at: Optional[int] = None):
"""
Perform file upload.
Performs continuous upload of chunks of the file. The size uploaded at each
cycle is the value of the attribute 'chunk_size'.
Parameters:
- stop_at (Optional[int]): Offset value where upload should stop.
Defaults to file size if not specified.
"""
def upload_chunk(self):
"""
Upload single chunk of file.
Uploads one chunk starting from current offset and updates offset
with server response.
"""
def create_url(self) -> str:
"""
Return new upload url from server.
Makes POST request to tus server to create a new upload url for the file.
Returns:
str: Upload URL for this file
Raises:
TusCommunicationError: If server returns error or no location header
"""Asynchronous Uploader
Asynchronous file uploader using the aiohttp library.
class AsyncUploader(BaseUploader):
"""Asynchronous file uploader using aiohttp library."""
async def upload(self, stop_at: Optional[int] = None):
"""
Perform file upload asynchronously.
Performs continuous upload of chunks of the file. The size uploaded at each
cycle is the value of the attribute 'chunk_size'.
Parameters:
- stop_at (Optional[int]): Offset value where upload should stop.
Defaults to file size if not specified.
"""
async def upload_chunk(self):
"""
Upload single chunk of file asynchronously.
Uploads one chunk starting from current offset and updates offset
with server response.
"""
async def create_url(self) -> str:
"""
Return new upload url from server asynchronously.
Makes async POST request to tus server to create a new upload url for the file.
Returns:
str: Upload URL for this file
Raises:
TusCommunicationError: If server returns error or no location header
"""Usage Examples
Basic Synchronous Upload
from tusclient import client
# Create client and uploader
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader('/path/to/file.ext', chunk_size=1024*1024)
# Upload entire file
uploader.upload()
# Upload chunk by chunk with progress tracking
while uploader.offset < uploader.get_file_size():
uploader.upload_chunk()
progress = (uploader.offset / uploader.get_file_size()) * 100
print(f"Upload progress: {progress:.1f}%")Basic Asynchronous Upload
import asyncio
from tusclient import client
async def upload_file():
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.async_uploader('/path/to/file.ext', chunk_size=1024*1024)
# Upload entire file
await uploader.upload()
# Run async upload
asyncio.run(upload_file())Resumable Upload with Storage
from tusclient import client
from tusclient.storage.filestorage import FileStorage
# Setup client with URL storage for resumability
storage = FileStorage('/tmp/tuspy_urls.db')
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader(
'/path/to/large_file.ext',
chunk_size=5*1024*1024, # 5MB chunks
store_url=True,
url_storage=storage,
retries=3,
retry_delay=10
)
try:
uploader.upload()
finally:
storage.close()Upload with Metadata
from tusclient import client
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader(
'/path/to/file.pdf',
metadata={
'filename': 'document.pdf',
'filetype': 'application/pdf',
'author': 'John Doe'
},
chunk_size=1024*1024
)
uploader.upload()Partial Upload
from tusclient import client
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader('/path/to/file.ext', chunk_size=1024*1024)
# Upload only first 50MB of file
uploader.upload(stop_at=50*1024*1024)Direct Uploader Usage
from tusclient.uploader import Uploader
# Use uploader without client if URL is known
uploader = Uploader(
'/path/to/file.ext',
url='http://tusd.tusdemo.net/files/abcdef123456',
chunk_size=1024*1024
)
uploader.upload()Install with Tessl CLI
npx tessl i tessl/pypi-tuspy