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

Session Management

Name: tessl/pypi-niquests
Author: tessl

Session classes for managing persistent connections, cookies, authentication, and other settings across multiple requests. Sessions provide connection pooling, configuration management, and state persistence, making them ideal for applications that make multiple requests to the same server or API.

Capabilities

Session Class

The synchronous Session class provides persistent connection management and configuration sharing across multiple HTTP requests.

class Session:
    """
    A Requests session for persistent configuration and connection management.
    
    Provides cookie persistence, connection-pooling, and configuration.
    Supports context manager protocol for automatic resource cleanup.
    """
    
    def __init__(
        self,
        *,
        retries: RetryType = DEFAULT_RETRIES,
        quic_cache_layer: CacheLayerAltSvcType | None = None,
        **kwargs
    ):
        """
        Initialize a new Session.
        
        Args:
            retries: Default retry configuration for all requests
            quic_cache_layer: QUIC connection cache layer
            **kwargs: Additional configuration options
        """
    
    def request(
        self,
        method: HttpMethodType,
        url: str,
        *,
        params: QueryParameterType | None = None,
        data: BodyType | None = None,
        json: Any | None = None,
        headers: HeadersType | None = None,
        cookies: CookiesType | None = None,
        files: MultiPartFilesType | MultiPartFilesAltType | None = None,
        auth: HttpAuthenticationType | None = None,
        timeout: TimeoutType | None = None,
        allow_redirects: bool = True,
        proxies: ProxyType | None = None,
        verify: TLSVerifyType | None = None,
        stream: bool = False,
        cert: TLSClientCertType | None = None,
        hooks: HookType[PreparedRequest | Response] | None = None,
    ) -> Response:
        """
        Send a request using the session configuration.
        
        Args:
            method: HTTP method
            url: Target URL
            **kwargs: Same parameters as module-level request functions
            
        Returns:
            Response object
        """
    
    def get(self, url: str, **kwargs) -> Response:
        """Send a GET request using session configuration."""
    
    def post(self, url: str, **kwargs) -> Response:
        """Send a POST request using session configuration."""
    
    def put(self, url: str, **kwargs) -> Response:
        """Send a PUT request using session configuration."""
    
    def patch(self, url: str, **kwargs) -> Response:
        """Send a PATCH request using session configuration."""
    
    def delete(self, url: str, **kwargs) -> Response:
        """Send a DELETE request using session configuration."""
    
    def head(self, url: str, **kwargs) -> Response:
        """Send a HEAD request using session configuration."""
    
    def options(self, url: str, **kwargs) -> Response:
        """Send an OPTIONS request using session configuration."""
    
    def prepare_request(self, request: Request) -> PreparedRequest:
        """
        Prepare a Request object for transmission.
        
        Args:
            request: Request object to prepare
            
        Returns:
            PreparedRequest ready for transmission
        """
    
    def send(
        self,
        request: PreparedRequest,
        *,
        stream: bool = False,
        timeout: TimeoutType | None = None,
        verify: TLSVerifyType | None = None,
        cert: TLSClientCertType | None = None,
        proxies: ProxyType | None = None,
        allow_redirects: bool = True,
    ) -> Response:
        """
        Send a PreparedRequest using session configuration.
        
        Args:
            request: PreparedRequest to send
            stream: Whether to stream response content
            timeout: Request timeout
            verify: SSL verification
            cert: Client certificate
            proxies: Proxy configuration
            allow_redirects: Follow redirects
            
        Returns:
            Response object
        """
    
    def close(self):
        """Close all underlying connections and clean up resources."""
    
    def __enter__(self) -> 'Session':
        """Enter context manager."""
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Exit context manager and clean up resources."""
        self.close()

AsyncSession Class

The asynchronous Session class provides the same functionality as Session but for async/await operations.

class AsyncSession(Session):
    """
    An asynchronous Requests session for concurrent request management.
    
    Inherits from Session and provides async versions of all methods.
    Supports async context manager protocol.
    """
    
    def __init__(
        self,
        *,
        retries: RetryType = DEFAULT_RETRIES,
        quic_cache_layer: CacheLayerAltSvcType | None = None,
        **kwargs
    ):
        """
        Initialize a new AsyncSession.
        
        Args:
            retries: Default retry configuration
            quic_cache_layer: QUIC connection cache layer
            **kwargs: Additional configuration options
        """
    
    async def request(
        self,
        method: HttpMethodType,
        url: str,
        *,
        params: QueryParameterType | None = None,
        data: AsyncBodyType | None = None,
        json: Any | None = None,
        headers: HeadersType | None = None,
        cookies: CookiesType | None = None,
        files: MultiPartFilesType | MultiPartFilesAltType | None = None,
        auth: AsyncHttpAuthenticationType | None = None,
        timeout: TimeoutType | None = None,
        allow_redirects: bool = True,
        proxies: ProxyType | None = None,
        verify: TLSVerifyType | None = None,
        stream: bool = False,
        cert: TLSClientCertType | None = None,
        hooks: HookType[PreparedRequest | AsyncResponse] | None = None,
    ) -> AsyncResponse:
        """
        Asynchronously send a request using session configuration.
        
        Args:
            method: HTTP method
            url: Target URL
            **kwargs: Same parameters as async request functions
            
        Returns:
            AsyncResponse object
        """
    
    async def get(self, url: str, **kwargs) -> AsyncResponse:
        """Asynchronously send a GET request."""
    
    async def post(self, url: str, **kwargs) -> AsyncResponse:
        """Asynchronously send a POST request."""
    
    async def put(self, url: str, **kwargs) -> AsyncResponse:
        """Asynchronously send a PUT request."""
    
    async def patch(self, url: str, **kwargs) -> AsyncResponse:
        """Asynchronously send a PATCH request."""
    
    async def delete(self, url: str, **kwargs) -> AsyncResponse:
        """Asynchronously send a DELETE request."""
    
    async def head(self, url: str, **kwargs) -> AsyncResponse:
        """Asynchronously send a HEAD request."""
    
    async def options(self, url: str, **kwargs) -> AsyncResponse:
        """Asynchronously send an OPTIONS request."""
    
    async def send(
        self,
        request: PreparedRequest,
        *,
        stream: bool = False,
        timeout: TimeoutType | None = None,
        verify: TLSVerifyType | None = None,
        cert: TLSClientCertType | None = None,
        proxies: ProxyType | None = None,
        allow_redirects: bool = True,
    ) -> AsyncResponse:
        """
        Asynchronously send a PreparedRequest.
        
        Args:
            request: PreparedRequest to send
            **kwargs: Same parameters as Session.send()
            
        Returns:
            AsyncResponse object
        """
    
    async def aclose(self):
        """Asynchronously close all connections and clean up resources."""
    
    def close(self):
        """Synchronously close connections (calls aclose() internally)."""
    
    async def __aenter__(self) -> 'AsyncSession':
        """Enter async context manager."""
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Exit async context manager and clean up resources."""
        await self.aclose()

Session Configuration

Both Session and AsyncSession classes maintain persistent configuration through instance attributes:

Core Session Attributes

class Session:
    # Request configuration
    headers: HeadersType  # Default headers for all requests
    cookies: RequestsCookieJar  # Cookie jar for automatic cookie handling
    auth: HttpAuthenticationType | None  # Default authentication
    proxies: ProxyType  # Default proxy configuration
    
    # SSL/TLS configuration
    verify: TLSVerifyType  # SSL certificate verification
    cert: TLSClientCertType | None  # Client certificate
    
    # Request behavior
    max_redirects: int  # Maximum number of redirects to follow
    trust_env: bool  # Whether to trust environment variables
    stream: bool  # Default streaming behavior
    
    # Connection management
    adapters: dict  # Protocol adapters (HTTP/HTTPS)
    
    # Hook configuration
    hooks: dict  # Request/response lifecycle hooks

Usage Examples

Basic Session Usage

import niquests

# Create a session with persistent configuration
session = niquests.Session()

# Set default headers that will be used for all requests
session.headers.update({
    'User-Agent': 'MyApp/1.0',
    'Accept': 'application/json'
})

# Set default authentication
session.auth = ('username', 'password')

# Make multiple requests with shared configuration
response1 = session.get('https://api.example.com/users')
response2 = session.get('https://api.example.com/posts')

# Session automatically handles cookies
login_response = session.post('https://api.example.com/login', {
    'username': 'user',
    'password': 'pass'
})

# Subsequent requests will include login cookies automatically
protected_response = session.get('https://api.example.com/protected')

# Clean up resources
session.close()

Context Manager Usage

# Recommended: Use as context manager for automatic cleanup
with niquests.Session() as session:
    session.headers['Authorization'] = 'Bearer token123'
    
    # Make requests using the configured session
    users = session.get('https://api.example.com/users').json()
    
    for user in users:
        profile = session.get(f'https://api.example.com/users/{user["id"]}')
        print(profile.json())

# Session is automatically closed when exiting the context

Async Session Usage

import asyncio
import niquests

async def fetch_data_async():
    async with niquests.AsyncSession() as session:
        # Configure session
        session.headers['Accept'] = 'application/json'
        session.timeout = 30.0
        
        # Make concurrent requests
        tasks = [
            session.get('https://api.example.com/endpoint1'),
            session.get('https://api.example.com/endpoint2'),
            session.get('https://api.example.com/endpoint3')
        ]
        
        responses = await asyncio.gather(*tasks)
        
        # Process responses
        results = []
        for response in responses:
            data = await response.json()
            results.append(data)
        
        return results

# Run async function
results = asyncio.run(fetch_data_async())

Advanced Session Configuration

from niquests import Session, TimeoutConfiguration, RetryConfiguration

# Create session with advanced configuration
session = Session(
    retries=RetryConfiguration(total=3, backoff_factor=0.5),
)

# Configure timeouts
session.timeout = TimeoutConfiguration(connect=5.0, read=30.0)

# Configure SSL verification
session.verify = '/path/to/ca-bundle.crt'
session.cert = ('/path/to/client.crt', '/path/to/client.key')

# Configure proxy
session.proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'https://proxy.example.com:8443'
}

# Add request hooks
def log_request(request, *args, **kwargs):
    print(f"Sending {request.method} request to {request.url}")

def log_response(response, *args, **kwargs):
    print(f"Received {response.status_code} response")

session.hooks = {
    'pre_request': [log_request],
    'response': [log_response]
}

# Use configured session
response = session.get('https://api.example.com/data')

Session vs Module-Level Functions

Use Sessions when:

Making multiple requests to the same server
Need to persist cookies between requests
Want to share configuration (headers, auth, proxies) across requests
Need connection pooling for better performance
Implementing authentication flows

Use module-level functions when:

Making single, standalone requests
Each request needs completely different configuration
Simple, one-off HTTP operations
Prototyping or testing

Performance Benefits

Sessions provide significant performance benefits:

Connection Pooling: Reuses underlying TCP connections
Cookie Management: Automatic cookie handling without manual intervention
Configuration Sharing: Reduces overhead of setting headers/auth per request
DNS Caching: Reduces DNS lookup overhead for repeated requests
SSL Session Reuse: Reuses SSL handshake results when possible

Type Definitions

CacheLayerAltSvcType = Any  # QUIC cache layer interface
RequestsCookieJar = CookieJar  # Cookie jar implementation

Install with Tessl CLI

npx tessl i tessl/pypi-niquests

docs