A Python client for the tus resumable upload protocol enabling pause and resume of file uploads
Custom exceptions for tus protocol communication errors and upload failures. TusPy provides specific exception types to help applications handle different error conditions appropriately.
Base exception for tus server communication errors.
class TusCommunicationError(Exception):
"""
Should be raised when communications with tus-server behaves unexpectedly.
Attributes:
message (str): Main message of the exception
status_code (int): Status code of response indicating an error
response_content (str): Content of response indicating an error
"""
def __init__(self, message: str, status_code: Optional[int] = None,
response_content: Optional[str] = None):
"""
Initialize TusCommunicationError.
Parameters:
- message (str): Error message. Can be None to generate default message using status_code
- status_code (Optional[int]): HTTP status code from server response
- response_content (Optional[str]): Content body from server response
"""Exception for failed upload attempts, inheriting from TusCommunicationError.
class TusUploadFailed(TusCommunicationError):
"""
Should be raised when an attempted upload fails.
Inherits all attributes and functionality from TusCommunicationError.
Used specifically for upload operation failures.
"""Decorator function for converting requests library exceptions to TusCommunicationError.
def catch_requests_error(func):
"""
Decorator to catch requests exceptions and convert to TusCommunicationError.
Wraps functions that make HTTP requests and converts any
requests.exceptions.RequestException to TusCommunicationError.
Parameters:
- func: Function to wrap
Returns:
Wrapped function that converts request exceptions
"""from tusclient import client
from tusclient.exceptions import TusCommunicationError, TusUploadFailed
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader('/path/to/file.ext', chunk_size=1024*1024)
try:
uploader.upload()
print("Upload completed successfully")
except TusUploadFailed as e:
print(f"Upload failed: {e}")
if e.status_code:
print(f"Server returned status code: {e.status_code}")
if e.response_content:
print(f"Server response: {e.response_content}")
except TusCommunicationError as e:
print(f"Communication error: {e}")
if e.status_code:
print(f"Status code: {e.status_code}")from tusclient import client
from tusclient.exceptions import TusCommunicationError, TusUploadFailed
def handle_upload_with_details(file_path: str, server_url: str):
my_client = client.TusClient(server_url)
uploader = my_client.uploader(file_path, chunk_size=1024*1024)
try:
uploader.upload()
return True, "Upload successful"
except TusUploadFailed as e:
error_details = {
'type': 'upload_failed',
'message': str(e),
'status_code': e.status_code,
'response_content': e.response_content
}
# Handle specific HTTP status codes
if e.status_code == 413:
error_details['user_message'] = "File too large for server"
elif e.status_code == 403:
error_details['user_message'] = "Upload not authorized"
elif e.status_code == 404:
error_details['user_message'] = "Upload URL not found - may have expired"
else:
error_details['user_message'] = f"Upload failed with status {e.status_code}"
return False, error_details
except TusCommunicationError as e:
error_details = {
'type': 'communication_error',
'message': str(e),
'status_code': e.status_code,
'user_message': "Server communication error"
}
return False, error_details
# Use the function
success, result = handle_upload_with_details('/path/to/file.ext', 'http://tusd.tusdemo.net/files/')
if success:
print(result)
else:
print(f"Error: {result['user_message']}")
print(f"Details: {result['message']}")import time
from tusclient import client
from tusclient.exceptions import TusCommunicationError, TusUploadFailed
def upload_with_custom_retry(file_path: str, server_url: str, max_retries: int = 3):
my_client = client.TusClient(server_url)
for attempt in range(max_retries + 1):
try:
uploader = my_client.uploader(file_path, chunk_size=1024*1024)
uploader.upload()
print(f"Upload successful on attempt {attempt + 1}")
return True
except TusUploadFailed as e:
# Don't retry on client errors (4xx)
if e.status_code and 400 <= e.status_code < 500:
print(f"Client error {e.status_code}: {e}")
return False
# Retry on server errors (5xx) or network issues
if attempt < max_retries:
wait_time = 2 ** attempt # Exponential backoff
print(f"Upload failed (attempt {attempt + 1}/{max_retries + 1}), retrying in {wait_time}s...")
time.sleep(wait_time)
else:
print(f"Upload failed after {max_retries + 1} attempts: {e}")
return False
except TusCommunicationError as e:
if attempt < max_retries:
wait_time = 2 ** attempt
print(f"Communication error (attempt {attempt + 1}/{max_retries + 1}), retrying in {wait_time}s...")
time.sleep(wait_time)
else:
print(f"Communication failed after {max_retries + 1} attempts: {e}")
return False
return False
# Use custom retry logic
success = upload_with_custom_retry('/path/to/file.ext', 'http://tusd.tusdemo.net/files/', max_retries=3)import asyncio
from tusclient import client
from tusclient.exceptions import TusCommunicationError, TusUploadFailed
async def async_upload_with_error_handling(file_path: str, server_url: str):
my_client = client.TusClient(server_url)
uploader = my_client.async_uploader(file_path, chunk_size=1024*1024)
try:
await uploader.upload()
print("Async upload completed successfully")
return True
except TusUploadFailed as e:
print(f"Async upload failed: {e}")
if e.status_code:
print(f"Status code: {e.status_code}")
return False
except TusCommunicationError as e:
print(f"Async communication error: {e}")
return False
# Run async upload
async def main():
success = await async_upload_with_error_handling('/path/to/file.ext', 'http://tusd.tusdemo.net/files/')
if success:
print("Upload operation completed")
else:
print("Upload operation failed")
asyncio.run(main())import logging
from tusclient import client
from tusclient.exceptions import TusCommunicationError, TusUploadFailed
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def upload_with_logging(file_path: str, server_url: str):
my_client = client.TusClient(server_url)
uploader = my_client.uploader(file_path, chunk_size=1024*1024)
try:
logger.info(f"Starting upload of {file_path}")
uploader.upload()
logger.info("Upload completed successfully")
return True
except TusUploadFailed as e:
logger.error(f"Upload failed: {e}")
logger.error(f"Status code: {e.status_code}")
logger.error(f"Response content: {e.response_content}")
return False
except TusCommunicationError as e:
logger.error(f"Communication error: {e}")
if e.status_code:
logger.error(f"Status code: {e.status_code}")
return False
except Exception as e:
logger.error(f"Unexpected error: {e}")
return False
# Use with logging
success = upload_with_logging('/path/to/file.ext', 'http://tusd.tusdemo.net/files/')Install with Tessl CLI
npx tessl i tessl/pypi-tuspy