Request Handling

class BaseTusRequest:
    """
    Http Request Abstraction.
    
    Sets up tus custom http request on instantiation.
    
    Attributes:
        response_headers (dict): Response headers from last request
        status_code (int): HTTP status code from last request
        response_content (bytes): Response content from last request
        verify_tls_cert (bool): Whether to verify TLS certificates
        file (IO): The file stream being uploaded
        client_cert (Union[str, Tuple[str, str]]): Client certificate configuration
    """
    
    def __init__(self, uploader):
        """
        Initialize BaseTusRequest with uploader configuration.
        
        Parameters:
        - uploader: Uploader instance providing configuration and file details
        """
    
    def add_checksum(self, chunk: bytes):
        """
        Add upload-checksum header for the provided chunk if enabled.
        
        Calculates checksum using configured algorithm and adds header
        in format required by tus protocol.
        
        Parameters:
        - chunk (bytes): File chunk data to calculate checksum for
        """

class TusRequest(BaseTusRequest):
    """Class to handle synchronous Tus upload requests using requests library."""
    
    def perform(self):
        """
        Perform actual HTTP PATCH request.
        
        Reads chunk data from file, calculates checksum if enabled,
        and sends PATCH request to upload URL with proper headers.
        
        Raises:
        TusUploadFailed: If request fails or returns error status
        """

class AsyncTusRequest(BaseTusRequest):
    """Class to handle async Tus upload requests using aiohttp library."""
    
    def __init__(self, *args, io_loop: Optional[asyncio.AbstractEventLoop] = None, **kwargs):
        """
        Initialize AsyncTusRequest with optional event loop.
        
        Parameters:
        - io_loop (Optional[asyncio.AbstractEventLoop]): Event loop for async operations
        - *args, **kwargs: Passed to BaseTusRequest constructor
        """
    
    async def perform(self):
        """
        Perform actual HTTP PATCH request asynchronously.
        
        Reads chunk data from file, calculates checksum if enabled,
        and sends async PATCH request to upload URL with proper headers
        and SSL configuration.
        
        Raises:
        TusUploadFailed: If request fails or returns error status
        """

def catch_requests_error(func):
    """
    Decorator to catch requests exceptions and convert to TusCommunicationError.
    
    Wraps functions that make HTTP requests using the requests library
    and converts any requests.exceptions.RequestException to TusCommunicationError
    for consistent error handling.
    
    Parameters:
    - func: Function to wrap that may raise requests exceptions
    
    Returns:
    Function: Wrapped function with error conversion
    """

from tusclient.request import TusRequest, AsyncTusRequest
from tusclient.uploader import Uploader, AsyncUploader
from tusclient import client

# Create uploader (request is normally created internally)
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader('/path/to/file.ext', chunk_size=1024*1024)

# Create and perform synchronous request manually
# (This is normally done automatically by uploader.upload_chunk())
request = TusRequest(uploader)
try:
    request.perform()
    print(f"Upload successful, offset: {request.response_headers.get('upload-offset')}")
    print(f"Status code: {request.status_code}")
except Exception as e:
    print(f"Request failed: {e}")

import asyncio
from tusclient.request import AsyncTusRequest
from tusclient import client

async def manual_async_request():
    my_client = client.TusClient('http://tusd.tusdemo.net/files/')
    uploader = my_client.async_uploader('/path/to/file.ext', chunk_size=1024*1024)
    
    # Create and perform async request manually
    # (This is normally done automatically by uploader.upload_chunk())
    request = AsyncTusRequest(uploader)
    try:
        await request.perform()
        print(f"Async upload successful, offset: {request.response_headers.get('upload-offset')}")
        print(f"Status code: {request.status_code}")
    except Exception as e:
        print(f"Async request failed: {e}")

# Run async request
asyncio.run(manual_async_request())

from tusclient.request import TusRequest, catch_requests_error
from tusclient import client
import requests

# Example of using catch_requests_error decorator
@catch_requests_error
def custom_tus_operation(url, headers):
    """Custom function that makes requests and converts exceptions."""
    response = requests.head(url, headers=headers)
    return response.headers.get('upload-offset')

try:
    my_client = client.TusClient('http://tusd.tusdemo.net/files/')
    headers = {'Tus-Resumable': '1.0.0'}
    offset = custom_tus_operation('http://tusd.tusdemo.net/files/some-id', headers)
    print(f"Current offset: {offset}")
except Exception as e:
    print(f"Custom operation failed: {e}")

from tusclient.request import TusRequest
from tusclient import client

# Example showing checksum handling (normally done automatically)
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader(
    '/path/to/file.ext', 
    chunk_size=1024*1024,
    upload_checksum=True  # Enable checksum validation
)

# Request will automatically include upload-checksum header
request = TusRequest(uploader)
chunk_data = b"example chunk data"
request.add_checksum(chunk_data)

# Check if checksum header was added
print("Request headers:", request._request_headers)
# Will show: upload-checksum: sha1 <base64-encoded-hash>

from tusclient.request import TusRequest
from tusclient import client

my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader('/path/to/file.ext', chunk_size=1024)

# Perform upload chunk to populate request
uploader.upload_chunk()

# Inspect request state (available after upload_chunk)
request = uploader.request
print(f"Last response status: {request.status_code}")
print(f"Response headers: {request.response_headers}")
print(f"Response content length: {len(request.response_content)}")
print(f"Upload offset: {request.response_headers.get('upload-offset')}")