Advanced Features

class TimeoutConfiguration:
    """
    Configuration for request timeouts.
    
    Provides detailed control over connection and read timeouts,
    allowing separate configuration for different phases of the request.
    """
    
    def __init__(
        self,
        connect: float | None = None,
        read: float | None = None,
        total: float | None = None,
        pool: float | None = None
    ):
        """
        Initialize timeout configuration.
        
        Args:
            connect: Timeout for establishing connection (seconds)
            read: Timeout for reading response data (seconds)
            total: Total timeout for entire request (seconds)
            pool: Timeout for getting connection from pool (seconds)
        """
    
    @property
    def connect_timeout(self) -> float | None:
        """Connection establishment timeout."""
    
    @property
    def read_timeout(self) -> float | None:
        """Response reading timeout."""

class RetryConfiguration:
    """
    Configuration for request retry behavior.
    
    Provides sophisticated retry logic with backoff strategies,
    status code filtering, and method-specific retry policies.
    """
    
    def __init__(
        self,
        total: int = 3,
        connect: int | None = None,
        read: int | None = None,
        redirect: int | None = None,
        status: int | None = None,
        other: int | None = None,
        allowed_methods: frozenset[str] | None = None,
        status_forcelist: frozenset[int] | None = None,
        backoff_factor: float = 0,
        backoff_max: float = 120,
        raise_on_redirect: bool = True,
        raise_on_status: bool = True,
        history: tuple | None = None,
        respect_retry_after_header: bool = True,
        remove_headers_on_redirect: frozenset[str] | None = None,
    ):
        """
        Initialize retry configuration.
        
        Args:
            total: Total number of retries
            connect: Retries for connection errors
            read: Retries for read errors
            redirect: Retries for redirect responses
            status: Retries for specific status codes
            other: Retries for other errors
            allowed_methods: HTTP methods that can be retried
            status_forcelist: HTTP status codes to force retry
            backoff_factor: Backoff multiplication factor
            backoff_max: Maximum backoff time
            raise_on_redirect: Raise exception on redirect failures
            raise_on_status: Raise exception on status failures
            history: Retry history tracking
            respect_retry_after_header: Honor Retry-After header
            remove_headers_on_redirect: Headers to remove on redirect
        """
    
    def new(self, **kwargs) -> RetryConfiguration:
        """
        Create new RetryConfiguration with updated parameters.
        
        Args:
            **kwargs: Parameters to update
            
        Returns:
            New RetryConfiguration instance
        """
    
    def get_backoff_time(self) -> float:
        """
        Calculate backoff time for next retry.
        
        Returns:
            Backoff time in seconds
        """

codes: LookupDict
"""
Dictionary-like object mapping HTTP status names to status codes.

Provides multiple access patterns for HTTP status codes:
- Attribute access: codes.ok, codes.not_found
- Dictionary access: codes['ok'], codes['not_found']
- Case-insensitive access: codes.OK, codes.Not_Found
- Special aliases: codes['\o/'] for 200

Common status codes:
- codes.ok = 200
- codes.created = 201
- codes.accepted = 202
- codes.no_content = 204
- codes.moved_permanently = 301
- codes.found = 302
- codes.not_modified = 304
- codes.temporary_redirect = 307
- codes.permanent_redirect = 308
- codes.bad_request = 400
- codes.unauthorized = 401
- codes.forbidden = 403
- codes.not_found = 404
- codes.method_not_allowed = 405
- codes.conflict = 409
- codes.gone = 410
- codes.unprocessable_entity = 422
- codes.too_many_requests = 429
- codes.internal_server_error = 500
- codes.bad_gateway = 502
- codes.service_unavailable = 503
- codes.gateway_timeout = 504
"""

class LookupDict(dict):
    """
    Dictionary-like object for status code lookups.
    
    Supports multiple access patterns and case-insensitive lookups.
    """
    
    def __getitem__(self, key: str | int) -> int:
        """Get status code by name or return key if it's already a code."""
    
    def get(self, key: str | int, default: int | None = None) -> int | None:
        """Get status code with default fallback."""

# URL and encoding utilities
def requote_uri(uri: str) -> str:
    """Re-quote the given URI with safe characters."""

def quote(s: str, safe: str = '') -> str:
    """Quote special characters in string."""

def unquote(s: str) -> str:
    """Unquote percent-encoded characters in string."""

def urldefrag(url: str) -> tuple[str, str]:
    """Remove fragment from URL."""

# Authentication utilities  
def get_auth_from_url(url: str) -> tuple[str, str] | None:
    """Extract authentication credentials from URL."""

def get_netrc_auth(url: str, raise_errors: bool = False) -> tuple[str, str] | None:
    """Get authentication from .netrc file."""

# Proxy utilities
def get_environ_proxies(url: str, no_proxy: str | None = None) -> dict[str, str]:
    """Get proxy configuration from environment variables."""

def should_bypass_proxies(url: str, no_proxy: str | None = None) -> bool:
    """Check if URL should bypass proxy configuration."""

def resolve_proxies(request: PreparedRequest, proxies: dict, trust_env: bool = True) -> dict:
    """Resolve proxy configuration for a request."""

# Header utilities
def default_headers() -> CaseInsensitiveDict:
    """Get default headers for requests."""

def default_user_agent(name: str = "niquests") -> str:
    """Get default User-Agent string."""

# Content utilities
def stream_decode_response_unicode(iterator: Iterator[bytes], encoding: str) -> Iterator[str]:
    """Decode response content as unicode stream."""

def get_encoding_from_headers(headers: HeadersType) -> str | None:
    """Extract encoding from Content-Type header."""

# Certificate and security utilities
def is_ocsp_capable() -> bool:
    """Check if OCSP certificate verification is available."""

def is_crl_capable() -> bool:
    """Check if CRL certificate verification is available."""

# Data structure utilities  
def to_key_val_list(value: dict | list | None) -> list[tuple[str, str]]:
    """Convert various data types to key-value pair list."""

HAS_LEGACY_URLLIB3: bool
"""
Boolean flag indicating if legacy urllib3 is being used.

This flag helps determine which features and behaviors are available
based on the urllib3 version. Some advanced features may not be
available with older urllib3 versions.

Usage:
    if HAS_LEGACY_URLLIB3:
        # Use legacy behavior
        pass
    else:
        # Use modern features
        pass
"""

__version__: str  # Package version (e.g., "3.15.2")
__title__: str  # Package title ("niquests")
__description__: str  # Package description
__url__: str  # Package homepage URL
__author__: str  # Author name
__author_email__: str  # Author email address
__license__: str  # License type ("Apache 2.0")
__copyright__: str  # Copyright notice
__build__: str  # Build information
__cake__: str  # Special build marker

import niquests
from niquests import TimeoutConfiguration

# Simple timeout (applies to both connect and read)
response = niquests.get('https://api.example.com/data', timeout=10.0)

# Separate connect and read timeouts
timeout_config = TimeoutConfiguration(connect=5.0, read=30.0)
response = niquests.get('https://api.example.com/data', timeout=timeout_config)

# Total timeout with separate phases
timeout_config = TimeoutConfiguration(
    connect=5.0,    # 5 seconds to establish connection
    read=30.0,      # 30 seconds to read response
    total=60.0      # Total request must complete within 60 seconds
)

# Use with session for persistent configuration
with niquests.Session() as session:
    session.timeout = timeout_config
    response1 = session.get('https://api.example.com/endpoint1')
    response2 = session.get('https://api.example.com/endpoint2')

import niquests
from niquests import RetryConfiguration

# Simple retry configuration
retry_config = RetryConfiguration(total=5)
response = niquests.get('https://api.example.com/data', retries=retry_config)

# Advanced retry with backoff
retry_config = RetryConfiguration(
    total=3,                    # Maximum 3 retries
    backoff_factor=0.5,         # Wait 0.5, 1.0, 2.0 seconds between retries
    backoff_max=10.0,           # Maximum 10 seconds backoff
    status_forcelist=[500, 502, 503, 504],  # Retry on server errors
    allowed_methods=['GET', 'POST'],         # Only retry safe methods
    respect_retry_after_header=True          # Honor server's Retry-After header
)

# Use with session
with niquests.Session() as session:
    session.retries = retry_config
    
    try:
        response = session.get('https://unreliable-api.example.com/data')
        print("Success after retries:", response.json())
    except niquests.RequestException as e:
        print("Failed after all retries:", e)

# Method-specific retry configuration
retry_config = RetryConfiguration(
    total=5,
    connect=2,      # Only 2 retries for connection errors
    read=3,         # 3 retries for read timeouts
    status=4        # 4 retries for HTTP status errors
)

import niquests

response = niquests.get('https://api.example.com/data')

# Using status code constants
if response.status_code == niquests.codes.ok:
    print("Request successful")
elif response.status_code == niquests.codes.not_found:
    print("Resource not found")
elif response.status_code == niquests.codes.unauthorized:
    print("Authentication required")
elif response.status_code >= niquests.codes.internal_server_error:
    print("Server error")

# Alternative access patterns
if response.status_code == niquests.codes['ok']:
    print("Success")

if response.status_code == niquests.codes.OK:  # Case insensitive
    print("Success")

# Status code ranges
def categorize_status(status_code):
    if 200 <= status_code < 300:
        return "success"
    elif 300 <= status_code < 400:
        return "redirect"
    elif 400 <= status_code < 500:
        return "client_error"
    elif 500 <= status_code < 600:
        return "server_error"
    else:
        return "unknown"

category = categorize_status(response.status_code)
print(f"Response category: {category}")

# Common status code checks
success_codes = [
    niquests.codes.ok,
    niquests.codes.created,
    niquests.codes.accepted,
    niquests.codes.no_content
]

if response.status_code in success_codes:
    print("Operation successful")

import niquests
from niquests import utils

# URL manipulation
original_url = "https://example.com/path with spaces"
safe_url = utils.requote_uri(original_url)
print(f"Safe URL: {safe_url}")

# Extract auth from URL
auth_url = "https://user:pass@api.example.com/data"
auth = utils.get_auth_from_url(auth_url)
if auth:
    username, password = auth
    print(f"Extracted auth: {username}")

# Environment proxy detection
url = "https://api.example.com/data"
proxies = utils.get_environ_proxies(url)
if proxies:
    print(f"Using proxies: {proxies}")

# Check if should bypass proxies
if utils.should_bypass_proxies(url):
    print("Bypassing proxy for this URL")

# Default headers
default_headers = utils.default_headers()
print(f"Default headers: {dict(default_headers)}")

# Custom User-Agent
user_agent = utils.default_user_agent("MyApp")
print(f"User-Agent: {user_agent}")

import niquests

# Check advanced security features
if niquests.is_ocsp_capable():
    print("OCSP certificate verification available")
    # Can use OCSP-based certificate verification
else:
    print("OCSP not available, using standard verification")

if niquests.is_crl_capable():
    print("CRL certificate verification available")
    # Can use Certificate Revocation Lists
else:
    print("CRL not available")

# Check urllib3 compatibility
if niquests.HAS_LEGACY_URLLIB3:
    print("Using legacy urllib3 - some features may be limited")
    # Adjust behavior for legacy compatibility
else:
    print("Using modern urllib3 - all features available")
    # Can use all advanced features

# Package information
print(f"Niquests version: {niquests.__version__}")
print(f"Build info: {niquests.__build__}")

import niquests
from niquests import TimeoutConfiguration, RetryConfiguration

# Comprehensive session configuration
class APISession(niquests.Session):
    """Custom session with advanced configuration."""
    
    def __init__(self, base_url, api_key=None):
        super().__init__()
        
        self.base_url = base_url
        
        # Configure timeouts
        self.timeout = TimeoutConfiguration(
            connect=5.0,
            read=30.0,
            total=60.0
        )
        
        # Configure retries
        self.retries = RetryConfiguration(
            total=3,
            backoff_factor=0.5,
            status_forcelist=[500, 502, 503, 504],
            allowed_methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE']
        )
        
        # Set default headers
        self.headers.update({
            'User-Agent': niquests.utils.default_user_agent('MyAPIClient/1.0'),
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        })
        
        # Set API key if provided
        if api_key:
            self.headers['Authorization'] = f'Bearer {api_key}'
    
    def request(self, method, endpoint, **kwargs):
        """Make request with automatic URL building."""
        url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
        return super().request(method, url, **kwargs)

# Usage
with APISession('https://api.example.com', api_key='secret') as session:
    # All requests will use the configured timeouts, retries, and headers
    users = session.get('/users').json()
    
    new_user = session.post('/users', json={
        'name': 'John Doe',
        'email': 'john@example.com'
    }).json()
    
    updated_user = session.put(f'/users/{new_user["id"]}', json={
        'name': 'John Smith'
    }).json()

import time
import niquests
from niquests import RetryConfiguration

def custom_retry_request(url, max_retries=3, backoff_base=2):
    """Custom retry logic with exponential backoff."""
    
    last_exception = None
    
    for attempt in range(max_retries + 1):
        try:
            response = niquests.get(url, timeout=10.0)
            
            # Check for retryable status codes
            if response.status_code in [500, 502, 503, 504]:
                if attempt < max_retries:
                    wait_time = backoff_base ** attempt
                    print(f"Server error {response.status_code}, retrying in {wait_time}s...")
                    time.sleep(wait_time)
                    continue
                else:
                    response.raise_for_status()
            
            # Success or non-retryable error
            return response
            
        except (niquests.ConnectionError, niquests.Timeout) as e:
            last_exception = e
            
            if attempt < max_retries:
                wait_time = backoff_base ** attempt
                print(f"Network error, retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise
    
    # If we get here, all retries were exhausted
    raise last_exception

# Usage
try:
    response = custom_retry_request('https://unreliable-api.example.com/data')
    print("Success:", response.json())
except niquests.RequestException as e:
    print(f"Failed after retries: {e}")