tessl/pypi-niquests

Niquests is a simple, yet elegant, HTTP library that is a drop-in replacement for Requests, which is under feature freeze.

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-niquests
Author: tessl

Comprehensive exception hierarchy for handling various error conditions during HTTP operations. These exceptions provide detailed error information and enable robust error handling patterns. All niquests exceptions inherit from RequestException, which inherits from Python's built-in IOError.

Exception Hierarchy

IOError
└── RequestException
    ├── HTTPError
    ├── ConnectionError
    │   ├── ProxyError
    │   ├── SSLError
    │   └── ConnectTimeout (also inherits from Timeout)
    ├── Timeout
    │   └── ReadTimeout
    ├── URLRequired
    ├── TooManyRedirects
    ├── InvalidJSONError
    │   └── JSONDecodeError
    ├── InvalidURL
    ├── InvalidSchema
    ├── InvalidHeader
    ├── MissingSchema
    ├── ChunkedEncodingError
    ├── ContentDecodingError
    ├── StreamConsumedError
    ├── RetryError
    ├── UnrewindableBodyError
    └── MultiplexingError

Capabilities

Base Exception Classes

Core exception classes that form the foundation of niquests error handling.

class RequestException(IOError):
    """
    Base exception class for all requests-related exceptions.
    
    There was an ambiguous exception that occurred while handling your request.
    All other niquests exceptions inherit from this class.
    """
    
    def __init__(self, *args, **kwargs):
        """
        Initialize RequestException with request and response objects.
        
        Args:
            *args: Exception arguments
            **kwargs: May include 'request' and 'response' objects
        """
    
    # Instance attributes
    request: PreparedRequest | None  # The request that caused the exception
    response: Response | None  # The response (if any) that caused the exception

class HTTPError(RequestException):
    """
    An HTTP error occurred.
    
    Raised when an HTTP request returns an unsuccessful status code
    (4xx or 5xx) and raise_for_status() is called.
    """

class ConnectionError(RequestException):
    """
    A Connection error occurred.
    
    Raised when there's a network-level connection problem,
    such as a refused connection or unreachable host.
    """

Timeout Exceptions

Exceptions related to request timeouts and timing issues.

class Timeout(RequestException):
    """
    The request timed out.
    
    Catching this error will catch both ConnectTimeout and ReadTimeout.
    This is the base class for all timeout-related exceptions.
    """

class ConnectTimeout(ConnectionError, Timeout):
    """
    The request timed out while trying to connect to the remote server.
    
    Requests that produced this error are safe to retry as the connection
    was never established.
    """

class ReadTimeout(Timeout):
    """
    The server did not send any data in the allotted amount of time.
    
    The connection was established but the server failed to send data
    within the specified timeout period.
    """

SSL and Security Exceptions

Exceptions related to SSL/TLS and security issues.

class SSLError(ConnectionError):
    """
    An SSL error occurred.
    
    Raised when there's an SSL/TLS-related problem such as:
    - Certificate verification failures
    - SSL handshake failures
    - Protocol version mismatches
    """

class ProxyError(ConnectionError):
    """
    A proxy error occurred.
    
    Raised when there's an issue with proxy configuration or
    the proxy server itself.
    """

URL and Schema Exceptions

Exceptions related to URL formatting and validation.

class URLRequired(RequestException):
    """
    A valid URL is required to make a request.
    
    Raised when a request is attempted without providing a URL.
    """

class InvalidURL(RequestException, ValueError):
    """
    The URL provided was somehow invalid.
    
    Raised when the URL cannot be parsed or contains invalid characters.
    """

class InvalidSchema(RequestException, ValueError):
    """
    The URL scheme provided is either invalid or unsupported.
    
    Raised for unsupported schemes like 'ftp://' or malformed schemes.
    """

class MissingSchema(RequestException, ValueError):
    """
    The URL scheme (e.g. http or https) is missing.
    
    Raised when a URL is provided without a protocol scheme.
    """

class InvalidHeader(RequestException, ValueError):
    """
    The header value provided was somehow invalid.
    
    Raised when HTTP headers contain invalid characters or formatting.
    """

Content and Encoding Exceptions

Exceptions related to response content processing and encoding.

class InvalidJSONError(RequestException):
    """
    A JSON error occurred.
    
    Base class for JSON-related errors during response processing.
    """

class JSONDecodeError(InvalidJSONError, json.JSONDecodeError):
    """
    Couldn't decode the text into json.
    
    Raised when response.json() is called but the response content
    is not valid JSON.
    """

class ChunkedEncodingError(RequestException):
    """
    The server declared chunked encoding but sent an invalid chunk.
    
    Raised when there's a problem with chunked transfer encoding
    in the response.
    """

class ContentDecodingError(RequestException):
    """
    Failed to decode response content.
    
    Raised when automatic content decompression (gzip, deflate, etc.)
    fails due to corrupted or invalid compressed data.
    """

class StreamConsumedError(RequestException, TypeError):
    """
    The content for this response was already consumed.
    
    Raised when attempting to read response content that has already
    been consumed by a previous operation.
    """

Advanced Operation Exceptions

Exceptions for advanced features and operations.

class TooManyRedirects(RequestException):
    """
    Too many redirects.
    
    Raised when the maximum number of redirects is exceeded.
    """

class RetryError(RequestException):
    """
    Custom retries logic failed.
    
    Raised when retry mechanisms are exhausted or encounter
    unrecoverable errors.
    """

class UnrewindableBodyError(RequestException):
    """
    Requests encountered an error when trying to rewind a body.
    
    Raised when attempting to retry a request with a body that
    cannot be rewound (e.g., a consumed stream).
    """

class MultiplexingError(RequestException):
    """
    Requests encountered an unresolvable error in multiplexed mode.
    
    Raised when HTTP/2 or HTTP/3 multiplexing encounters errors
    that cannot be resolved.
    """

Warning Classes

Warning classes for non-fatal issues and deprecation notices.

class RequestsWarning(Warning):
    """
    Base warning for Requests.
    
    Base class for all niquests-related warnings.
    """

class FileModeWarning(RequestsWarning, DeprecationWarning):
    """
    A file was opened in text mode, but Requests determined its binary length.
    
    Warning raised when files are opened in text mode but should be
    opened in binary mode for proper handling.
    """

class RequestsDependencyWarning(RequestsWarning):
    """
    An imported dependency doesn't match the expected version range.
    
    Warning raised when dependency versions may cause compatibility issues.
    """

Usage Examples

Basic Exception Handling

import niquests

try:
    response = niquests.get('https://api.example.com/data', timeout=5.0)
    response.raise_for_status()  # Raise HTTPError for bad status codes
    data = response.json()
    print("Success:", data)
    
except niquests.HTTPError as e:
    print(f"HTTP Error {e.response.status_code}: {e.response.reason}")
    # Access the response even when there's an error
    if e.response.headers.get('content-type', '').startswith('application/json'):
        error_details = e.response.json()
        print(f"Error details: {error_details}")
        
except niquests.ConnectionError:
    print("Failed to connect to the server")
    
except niquests.Timeout:
    print("Request timed out")
    
except niquests.RequestException as e:
    print(f"An error occurred: {e}")

Specific Exception Handling

import niquests

def robust_api_call(url, max_retries=3):
    """Make API call with comprehensive error handling."""
    
    for attempt in range(max_retries):
        try:
            response = niquests.get(url, timeout=(5.0, 30.0))
            response.raise_for_status()
            return response.json()
            
        except niquests.ConnectTimeout:
            print(f"Connection timeout on attempt {attempt + 1}")
            if attempt == max_retries - 1:
                raise
                
        except niquests.ReadTimeout:
            print(f"Read timeout on attempt {attempt + 1}")
            if attempt == max_retries - 1:
                raise
                
        except niquests.SSLError as e:
            print(f"SSL error: {e}")
            # SSL errors are usually not retryable
            raise
            
        except niquests.HTTPError as e:
            status_code = e.response.status_code
            if status_code >= 500:
                # Server errors might be retryable
                print(f"Server error {status_code} on attempt {attempt + 1}")
                if attempt == max_retries - 1:
                    raise
            else:
                # Client errors (4xx) are usually not retryable
                print(f"Client error {status_code}: {e.response.reason}")
                raise
                
        except niquests.JSONDecodeError:
            print("Response is not valid JSON")
            print(f"Response content: {response.text[:200]}...")
            raise
            
        except niquests.TooManyRedirects:
            print("Too many redirects - possible redirect loop")
            raise
            
        # Wait before retry
        if attempt < max_retries - 1:
            time.sleep(2 ** attempt)  # Exponential backoff
    
    raise niquests.RequestException("Max retries exceeded")

# Usage
try:
    data = robust_api_call('https://api.example.com/data')
    print("API call successful:", data)
except niquests.RequestException as e:
    print(f"API call failed: {e}")

URL Validation and Error Handling

import niquests

def validate_and_fetch(url):
    """Validate URL and fetch content with proper error handling."""
    
    try:
        # This will raise various URL-related exceptions
        response = niquests.get(url)
        return response.text
        
    except niquests.MissingSchema:
        print("URL is missing a scheme (http:// or https://)")
        # Try to fix by adding https://
        return validate_and_fetch(f"https://{url}")
        
    except niquests.InvalidSchema as e:
        print(f"Invalid or unsupported URL scheme: {e}")
        raise
        
    except niquests.InvalidURL as e:
        print(f"Malformed URL: {e}")
        raise
        
    except niquests.ConnectionError:
        print("Could not connect to the server")
        raise

# Usage examples
try:
    content = validate_and_fetch("example.com")  # Missing scheme
    print("Content fetched successfully")
except niquests.RequestException as e:
    print(f"Failed to fetch content: {e}")

Async Exception Handling

import asyncio
import niquests

async def async_request_with_error_handling(url):
    """Async request with comprehensive error handling."""
    
    try:
        response = await niquests.aget(url, timeout=10.0)
        
        async with response:  # Ensure response is closed
            response.raise_for_status()
            data = await response.json()
            return data
            
    except niquests.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code}")
        # Can still access response data
        error_content = await e.response.text
        print(f"Error response: {error_content}")
        raise
        
    except niquests.ConnectionError:
        print("Connection failed")
        raise
        
    except niquests.Timeout:
        print("Request timed out")
        raise
        
    except niquests.JSONDecodeError:
        print("Invalid JSON response")
        raise

# Usage
async def main():
    try:
        data = await async_request_with_error_handling('https://api.example.com/data')
        print("Success:", data)
    except niquests.RequestException as e:
        print(f"Request failed: {e}")

asyncio.run(main())

Custom Exception Handling

import niquests
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class APIClient:
    """Example API client with comprehensive error handling."""
    
    def __init__(self, base_url, timeout=30.0):
        self.base_url = base_url
        self.timeout = timeout
        self.session = niquests.Session()
    
    def _handle_request_error(self, e, url):
        """Centralized error handling and logging."""
        
        if isinstance(e, niquests.HTTPError):
            status = e.response.status_code
            logger.error(f"HTTP {status} error for {url}: {e.response.reason}")
            
            # Log response content for debugging
            try:
                error_data = e.response.json()
                logger.error(f"Error response: {error_data}")
            except niquests.JSONDecodeError:
                logger.error(f"Error response (non-JSON): {e.response.text[:500]}")
                
        elif isinstance(e, niquests.ConnectionError):
            logger.error(f"Connection error for {url}: {e}")
            
        elif isinstance(e, niquests.Timeout):
            logger.error(f"Timeout error for {url}: {e}")
            
        elif isinstance(e, niquests.JSONDecodeError):
            logger.error(f"JSON decode error for {url}: {e}")
            
        else:
            logger.error(f"Unexpected error for {url}: {e}")
    
    def get(self, endpoint, **kwargs):
        """Make GET request with error handling."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        try:
            response = self.session.get(url, timeout=self.timeout, **kwargs)
            response.raise_for_status()
            return response.json()
            
        except niquests.RequestException as e:
            self._handle_request_error(e, url)
            raise
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.session.close()

# Usage
with APIClient('https://api.example.com') as client:
    try:
        users = client.get('/users')
        print(f"Found {len(users)} users")
    except niquests.RequestException:
        print("Failed to fetch users")

Best Practices

Exception Handling Strategy

Catch specific exceptions rather than using broad except clauses
Handle retryable vs non-retryable errors differently
Log errors appropriately with relevant context
Preserve error information for debugging
Use response data even for errors when available

Common Patterns

# Pattern 1: Catch and retry
def retry_request(url, max_retries=3):
    for i in range(max_retries):
        try:
            return niquests.get(url, timeout=10.0)
        except (niquests.ConnectionError, niquests.Timeout):
            if i == max_retries - 1:
                raise
            time.sleep(2 ** i)

# Pattern 2: Graceful degradation
def get_user_data(user_id, fallback=None):
    try:
        response = niquests.get(f'/api/users/{user_id}')
        response.raise_for_status()
        return response.json()
    except niquests.RequestException:
        logger.warning(f"Failed to fetch user {user_id}, using fallback")
        return fallback or {'id': user_id, 'name': 'Unknown User'}

# Pattern 3: Error categorization
def categorize_error(exception):
    if isinstance(exception, niquests.HTTPError):
        status = exception.response.status_code
        if 400 <= status < 500:
            return 'client_error'
        elif 500 <= status < 600:
            return 'server_error'
    elif isinstance(exception, niquests.ConnectionError):
        return 'network_error'
    elif isinstance(exception, niquests.Timeout):
        return 'timeout_error'
    return 'unknown_error'

Install with Tessl CLI