tessl/pypi-treq

High-level Twisted HTTP Client API for asynchronous HTTP requests in Python

Overview

Eval results

Files

HTTP Client Class

Name: tessl/pypi-treq
Author: tessl

Advanced HTTP client providing direct access to connection pooling, custom agents, and request lifecycle management for applications requiring fine-grained control over HTTP behavior.

Capabilities

HTTPClient Class

Core HTTP client implementation that wraps Twisted's Agent system with cookie management and request processing.

class HTTPClient:
    def __init__(self, agent, cookiejar=None, data_to_body_producer=IBodyProducer):
        """
        Initialize HTTP client with custom agent and configuration.
        
        Parameters:
        - agent: IAgent - Twisted agent for making HTTP requests
        - cookiejar: CookieJar or None - Cookie jar for session management
        - data_to_body_producer: callable - Function to convert data to IBodyProducer
        """
    
    def request(self, method, url, **kwargs):
        """
        Make an HTTP request with specified method.
        
        Parameters:
        - method: str - HTTP method
        - url: str, bytes, EncodedURL, or DecodedURL - Target URL
        - **kwargs: Same parameters as module-level request functions
        
        Returns:
        Deferred that fires with _Response object
        """
    
    def get(self, url, **kwargs):
        """Make a GET request. See module-level get() function."""
    
    def post(self, url, data=None, **kwargs):
        """Make a POST request. See module-level post() function."""
    
    def put(self, url, data=None, **kwargs):
        """Make a PUT request. See module-level put() function."""
    
    def patch(self, url, data=None, **kwargs):
        """Make a PATCH request. See module-level patch() function."""
    
    def delete(self, url, **kwargs):
        """Make a DELETE request. See module-level delete() function."""
    
    def head(self, url, **kwargs):
        """Make a HEAD request. See module-level head() function."""

Connection Pool Management

Functions for managing HTTP connection pools across requests for improved performance.

def get_global_pool():
    """
    Get the current global connection pool.
    
    Returns:
    HTTPConnectionPool or None
    """

def set_global_pool(pool):
    """
    Set the global connection pool.
    
    Parameters:
    - pool: HTTPConnectionPool - Connection pool to set as global
    """

def default_pool(reactor, pool, persistent):
    """
    Get appropriate connection pool based on parameters.
    
    Parameters:
    - reactor: Twisted reactor
    - pool: HTTPConnectionPool or None - Explicit pool
    - persistent: bool or None - Use persistent connections
    
    Returns:
    HTTPConnectionPool - Connection pool to use
    """

Reactor Management

Utility for reactor selection and configuration.

def default_reactor(reactor):
    """
    Return the specified reactor or import the default reactor.
    
    Parameters:
    - reactor: Twisted reactor or None
    
    Returns:
    Twisted reactor instance
    """

Usage Examples

Custom HTTP Client

from twisted.web.client import Agent, HTTPConnectionPool
from treq.client import HTTPClient
from twisted.internet import reactor
from http.cookiejar import CookieJar

# Create custom agent with specific configuration
pool = HTTPConnectionPool(reactor, persistent=True)
agent = Agent(reactor, pool=pool, connectTimeout=30)

# Create client with custom cookie jar
cookiejar = CookieJar()
client = HTTPClient(agent, cookiejar=cookiejar)

# Use the client
@defer.inlineCallbacks
def use_custom_client():
    response = yield client.get('https://httpbin.org/cookies/set/test/value')
    cookies_response = yield client.get('https://httpbin.org/cookies')
    data = yield cookies_response.json()
    print("Cookies:", data['cookies'])

Connection Pool Configuration

from twisted.web.client import HTTPConnectionPool
from treq.client import set_global_pool
from twisted.internet import reactor

# Configure global connection pool
pool = HTTPConnectionPool(
    reactor,
    persistent=True,
    maxPersistentPerHost=10,
    cachedConnectionTimeout=240,
    retryAutomatically=True
)
set_global_pool(pool)

# Now all treq requests will use this pool
import treq

@defer.inlineCallbacks
def use_configured_pool():
    # This request uses the custom pool configuration
    response = yield treq.get('https://httpbin.org/get')

Advanced Agent Configuration

from twisted.web.client import Agent, BrowserLikePolicyForHTTPS
from twisted.internet.ssl import ClientContextFactory
from treq.client import HTTPClient
import treq

# Custom SSL context
class CustomContextFactory(ClientContextFactory):
    def getContext(self, hostname, port):
        context = ClientContextFactory.getContext(self, hostname, port)
        # Configure SSL context as needed
        return context

# Create agent with custom SSL and policy
agent = Agent(
    reactor,
    contextFactory=CustomContextFactory(),
    pool=HTTPConnectionPool(reactor, persistent=True),
    connectTimeout=30
)

client = HTTPClient(agent)

@defer.inlineCallbacks
def secure_request():
    response = yield client.get('https://secure-api.example.com/data')
    data = yield response.json()

Custom Body Producer

from twisted.web.iweb import IBodyProducer
from zope.interface import implementer
from twisted.internet import defer

@implementer(IBodyProducer)
class CustomBodyProducer:
    def __init__(self, data):
        self.data = data
        self.length = len(data)
    
    def startProducing(self, consumer):
        consumer.write(self.data)
        return defer.succeed(None)
    
    def pauseProducing(self):
        pass
    
    def stopProducing(self):
        pass

# Custom data-to-body-producer function
def custom_data_converter(data):
    if isinstance(data, dict) and 'custom' in data:
        return CustomBodyProducer(data['custom'].encode('utf-8'))
    return IBodyProducer(data)  # Use default conversion

# Client with custom body producer
client = HTTPClient(agent, data_to_body_producer=custom_data_converter)

@defer.inlineCallbacks
def use_custom_producer():
    response = yield client.post(
        'https://httpbin.org/post',
        data={'custom': 'special data format'}
    )

Session Management

from http.cookiejar import CookieJar
from treq.client import HTTPClient
from twisted.web.client import Agent

# Create client with persistent cookies
cookiejar = CookieJar()
client = HTTPClient(Agent(reactor), cookiejar=cookiejar)

@defer.inlineCallbacks
def session_example():
    # Login request - cookies are automatically stored
    login_response = yield client.post(
        'https://httpbin.org/cookies/set/session/abc123',
        data={'username': 'user', 'password': 'pass'}
    )
    
    # Subsequent requests automatically include cookies
    profile_response = yield client.get('https://httpbin.org/cookies')
    profile_data = yield profile_response.json()
    print("Session cookies:", profile_data['cookies'])
    
    # Logout
    yield client.post('https://httpbin.org/cookies/delete/session')

Error Handling and Retries

from twisted.internet import defer
from twisted.web.client import ResponseFailed

@defer.inlineCallbacks
def robust_client():
    client = HTTPClient(Agent(reactor))
    
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = yield client.get('https://unreliable-api.example.com/data')
            data = yield response.json()
            defer.returnValue(data)
        except ResponseFailed as e:
            if attempt == max_retries - 1:
                raise
            print(f"Attempt {attempt + 1} failed, retrying...")
            yield defer.sleep(2 ** attempt)  # Exponential backoff

Types

HTTP client related types:

# Agent interface from Twisted
from twisted.web.iweb import IAgent

# Cookie jar types
from http.cookiejar import CookieJar

# Body producer interface
from twisted.web.iweb import IBodyProducer

# Connection pool
from twisted.web.client import HTTPConnectionPool

# Data to body producer converter
DataToBodyProducer = Callable[[Any], IBodyProducer]

# Response type
from treq.response import _Response

Performance Optimization

Connection Pooling

Persistent connections: Enable persistent=True for connection reuse
Pool sizing: Configure maxPersistentPerHost based on expected load
Timeout management: Set appropriate cachedConnectionTimeout values
Global pools: Use set_global_pool() to share pools across application

Memory Management

Cookie jar cleanup: Periodically clear unnecessary cookies
Response buffering: Use unbuffered=True for large responses
Connection limits: Set reasonable connection pool limits
Reactor sharing: Reuse reactor instances across clients

Security Considerations

SSL verification: Configure appropriate SSL context factories
Certificate validation: Use BrowserLikePolicyForHTTPS for proper validation
Cookie security: Handle secure and HttpOnly cookies appropriately
Timeout configuration: Set reasonable timeouts to prevent hanging connections