tessl/pypi-http3

A next generation HTTP client for Python 3 with HTTP/2 support, async/await capabilities, and requests-compatible API.

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-http3
Author: tessl

Comprehensive exception hierarchy for handling different types of HTTP errors, timeouts, connection issues, and protocol violations. HTTP3 provides specific exception types that enable precise error handling and appropriate recovery strategies.

Capabilities

Timeout Exceptions

Exceptions related to various timeout scenarios during HTTP operations.

class Timeout(Exception):
    """
    Base class for all timeout exceptions.
    
    All timeout-related exceptions inherit from this class, allowing
    you to catch any timeout condition with a single except block.
    """

class ConnectTimeout(Timeout):
    """
    Timeout while establishing a connection.
    
    Raised when the client cannot establish a TCP connection to the
    server within the specified connect_timeout period.
    """

class ReadTimeout(Timeout):
    """
    Timeout while reading response data.
    
    Raised when the server doesn't send response data within the
    specified read_timeout period after connection is established.
    """

class WriteTimeout(Timeout):
    """
    Timeout while writing request data.
    
    Raised when the client cannot send request data to the server
    within the specified write_timeout period.
    """

class PoolTimeout(Timeout):
    """
    Timeout while waiting to acquire a connection from the pool.
    
    Raised when no connection becomes available in the connection
    pool within the specified pool_timeout period.
    """

Usage Example:

import http3

try:
    response = http3.get('https://slow-server.example.com', timeout=5.0)
except http3.ConnectTimeout:
    print("Could not connect to server")
except http3.ReadTimeout:
    print("Server didn't respond in time")
except http3.WriteTimeout:
    print("Could not send request data")
except http3.PoolTimeout:
    print("No connection available in pool")
except http3.Timeout:
    print("Some kind of timeout occurred")

HTTP Protocol Exceptions

Exceptions related to HTTP protocol violations and general HTTP errors.

class ProtocolError(Exception):
    """
    Malformed HTTP protocol data.
    
    Raised when the server sends invalid HTTP data that violates
    the HTTP specification, such as malformed headers or invalid
    response structure.
    """

class DecodingError(Exception):
    """
    Content decoding failure.
    
    Raised when response content cannot be decoded properly,
    such as invalid UTF-8 encoding or corrupted compressed data.
    """

class InvalidURL(Exception):
    """
    Invalid URL format.
    
    Raised when a provided URL is malformed, missing required
    components (scheme, host), or contains invalid characters.
    """

Usage Example:

import http3

try:
    response = http3.get('invalid://url')
except http3.InvalidURL:
    print("URL format is invalid")

try:
    response = http3.get('https://api.example.com/data')
    data = response.json()
except http3.ProtocolError:
    print("Server sent malformed HTTP data")
except http3.DecodingError:
    print("Could not decode response content")

Redirect Exceptions

Exceptions related to HTTP redirect handling and redirect loops.

class TooManyRedirects(Exception):
    """
    Too many redirects occurred.
    
    Raised when the number of redirects exceeds the maximum
    allowed limit (typically 20 by default).
    """

class RedirectBodyUnavailable(Exception):
    """
    Redirect attempted but request body was streaming and is no longer available.
    
    Raised when a POST/PUT request with streaming body encounters
    a redirect, but the body cannot be replayed for the redirect.
    """

class RedirectLoop(Exception):
    """
    Infinite redirect loop detected.
    
    Raised when the client detects that redirects are forming
    a loop, visiting the same URLs repeatedly.
    """

Usage Example:

import http3

try:
    response = http3.get('https://example.com/redirect-loop')
except http3.TooManyRedirects:
    print("Too many redirects - possible redirect loop")
except http3.RedirectLoop:
    print("Infinite redirect loop detected")

try:
    with open('large-file.bin', 'rb') as f:
        response = http3.post('https://example.com/upload', data=f)
except http3.RedirectBodyUnavailable:
    print("Cannot redirect streaming request body")

Stream Exceptions

Exceptions related to response streaming and content access patterns.

class StreamConsumed(Exception):
    """
    Attempted to read or stream response content that has already been consumed.
    
    Raised when trying to access response content after it has
    already been read through streaming or direct access.
    """

class ResponseNotRead(Exception):
    """
    Attempted to access response content without reading it first.
    
    Raised when trying to access response content properties
    after a streaming response that hasn't been fully read.
    """

class ResponseClosed(Exception):
    """
    Attempted to read or stream response content from a closed response.
    
    Raised when trying to access content from a response that
    has been explicitly closed or whose connection was terminated.
    """

Usage Example:

import http3

try:
    response = http3.get('https://api.example.com/data', stream=True)
    
    # Consume the stream
    for chunk in response.iter_content():
        process_chunk(chunk)
    
    # This will raise StreamConsumed
    text = response.text
    
except http3.StreamConsumed:
    print("Response content already consumed")
except http3.ResponseNotRead:
    print("Response not fully read")
except http3.ResponseClosed:
    print("Response connection closed")

Connection Exceptions

Exceptions related to connection management and network issues.

class NotConnected(Exception):
    """
    Connection was lost at the point of starting a request.
    
    Raised when a connection is lost before any request data
    has been successfully sent.
    """

class CookieConflict(Exception):
    """
    Attempted to lookup a cookie by name, but multiple cookies existed.
    
    Raised when trying to access a cookie by name when multiple
    cookies with the same name exist for different domains/paths.
    """

Error Handling Strategies

Comprehensive Error Handling

import http3
import time

def robust_request(url, max_retries=3, backoff_factor=1.0):
    """Make a request with comprehensive error handling and retries."""
    
    for attempt in range(max_retries):
        try:
            response = http3.get(url, timeout=30.0)
            response.raise_for_status()  # Raise for HTTP error status codes
            return response
            
        except http3.ConnectTimeout:
            print(f"Connection timeout (attempt {attempt + 1})")
            if attempt < max_retries - 1:
                time.sleep(backoff_factor * (2 ** attempt))
                continue
            raise
            
        except http3.ReadTimeout:
            print(f"Read timeout (attempt {attempt + 1})")
            if attempt < max_retries - 1:
                time.sleep(backoff_factor * (2 ** attempt))
                continue
            raise
            
        except http3.ProtocolError as e:
            print(f"Protocol error: {e}")
            # Don't retry protocol errors
            raise
            
        except http3.TooManyRedirects:
            print("Too many redirects")
            # Don't retry redirect errors
            raise
            
        except http3.InvalidURL as e:
            print(f"Invalid URL: {e}")
            # Don't retry URL errors
            raise
            
        except Exception as e:
            print(f"Unexpected error: {e}")
            if attempt < max_retries - 1:
                time.sleep(backoff_factor * (2 ** attempt))
                continue
            raise

# Usage
try:
    response = robust_request('https://api.example.com/data')
    data = response.json()
except http3.Timeout:
    print("Request timed out after retries")
except http3.ProtocolError:
    print("Server sent invalid HTTP data")
except Exception as e:
    print(f"Request failed: {e}")

Async Error Handling

import http3
import asyncio

async def async_request_with_handling(url):
    """Async request with proper error handling."""
    
    try:
        async with http3.AsyncClient(timeout=30.0) as client:
            response = await client.get(url)
            
            # Check for HTTP errors
            if response.is_error:
                print(f"HTTP Error: {response.status_code}")
                return None
            
            return await response.json()
            
    except http3.ConnectTimeout:
        print("Could not connect to server")
    except http3.ReadTimeout:
        print("Server response timeout")
    except http3.ProtocolError as e:
        print(f"Protocol error: {e}")
    except http3.DecodingError as e:
        print(f"Content decoding error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")
    
    return None

# Usage
async def main():
    data = await async_request_with_handling('https://api.example.com/data')
    if data:
        print(f"Received data: {data}")

asyncio.run(main())

Context-Specific Error Handling

import http3

def handle_api_request(url, api_key):
    """Handle API requests with context-specific error handling."""
    
    try:
        response = http3.get(
            url,
            headers={'Authorization': f'Bearer {api_key}'},
            timeout=10.0
        )
        
        # Handle HTTP status codes
        if response.status_code == 401:
            raise AuthenticationError("Invalid API key")
        elif response.status_code == 429:
            raise RateLimitError("Rate limit exceeded")
        elif response.status_code >= 500:
            raise ServerError(f"Server error: {response.status_code}")
        
        response.raise_for_status()
        return response.json()
        
    except http3.ConnectTimeout:
        raise NetworkError("Cannot connect to API server")
    except http3.ReadTimeout:
        raise NetworkError("API server response timeout")
    except http3.InvalidURL:
        raise ConfigurationError("Invalid API URL")
    except http3.DecodingError:
        raise DataError("Invalid response format")

# Custom exception hierarchy
class APIError(Exception):
    """Base API error."""

class NetworkError(APIError):
    """Network-related error."""

class AuthenticationError(APIError):
    """Authentication error."""

class RateLimitError(APIError):
    """Rate limit error."""

class ServerError(APIError):
    """Server error."""

class ConfigurationError(APIError):
    """Configuration error."""

class DataError(APIError):
    """Data processing error."""

Response Streaming Error Handling

import http3

def download_file_with_error_handling(url, filename):
    """Download a file with proper streaming error handling."""
    
    try:
        response = http3.get(url, stream=True)
        response.raise_for_status()
        
        with open(filename, 'wb') as f:
            try:
                for chunk in response.iter_content(chunk_size=8192):
                    f.write(chunk)
                    
            except http3.ReadTimeout:
                print("Download interrupted - read timeout")
                response.close()
                # Remove partial file
                import os
                if os.path.exists(filename):
                    os.remove(filename)
                raise
                
            except http3.StreamConsumed:
                print("Stream already consumed")
                raise
                
            except http3.ResponseClosed:
                print("Connection closed during download")
                raise
                
        print(f"Downloaded {filename} successfully")
        
    except http3.ConnectTimeout:
        print("Could not connect to download server")
    except http3.ProtocolError:
        print("Invalid response from server")
    except Exception as e:
        print(f"Download failed: {e}")
        # Clean up partial file
        import os
        if os.path.exists(filename):
            os.remove(filename)
        raise

# Usage
try:
    download_file_with_error_handling(
        'https://example.com/large-file.zip',
        'downloaded-file.zip'
    )
except Exception:
    print("Download failed")

Exception Hierarchy Summary

Exception
├── Timeout
│   ├── ConnectTimeout
│   ├── ReadTimeout
│   ├── WriteTimeout
│   └── PoolTimeout
├── ProtocolError
├── DecodingError
├── InvalidURL
├── TooManyRedirects
├── RedirectBodyUnavailable
├── RedirectLoop
├── StreamConsumed
├── ResponseNotRead
├── ResponseClosed
├── NotConnected
└── CookieConflict

This hierarchy allows for both specific and general exception handling patterns depending on your application's needs.

Install with Tessl CLI