tessl/pypi-twisted

An asynchronous networking framework written in Python

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

HTTP Client and Server

Name: tessl/pypi-twisted
Author: tessl

Complete HTTP implementation with client agents, server resources, and web application framework. Twisted's web module provides both high-level and low-level APIs for HTTP operations, WSGI support, and extensive web service capabilities.

Capabilities

HTTP Client

HTTP client functionality with support for connection pooling, redirects, cookies, content encoding, and TLS.

class client.Agent:
    """
    HTTP client agent for making requests.
    """
    def __init__(self, reactor, contextFactory=None, connectTimeout=None, bindAddress=None, pool=None):
        """
        Args:
            reactor: The reactor to use
            contextFactory: SSL context factory for HTTPS
            connectTimeout (float): Connection timeout in seconds
            bindAddress: Local address to bind connections
            pool: Connection pool to use
        """
    
    def request(self, method, uri, headers=None, bodyProducer=None):
        """
        Make an HTTP request.
        
        Args:
            method (bytes): HTTP method (b'GET', b'POST', etc.)
            uri (bytes): Request URI
            headers (Headers): HTTP headers
            bodyProducer: Body producer for request data
            
        Returns:
            Deferred[Response]: Fires with Response object
        """

class client.Response:
    """
    HTTP response object.
    
    Attributes:
    - version: HTTP version tuple (major, minor)
    - code: HTTP status code (int)
    - phrase: HTTP status phrase (bytes)
    - headers: Response headers (Headers object)
    - length: Content length or UNKNOWN_LENGTH
    """
    version = None
    code = None
    phrase = None
    headers = None
    length = None
    
    def deliverBody(self, protocol):
        """
        Deliver response body to a protocol.
        
        Args:
            protocol: Protocol to receive body data
        """

def client.readBody(response):
    """
    Read the entire body of an HTTP response.
    
    Args:
        response (Response): HTTP response object
        
    Returns:
        Deferred[bytes]: Complete response body
    """

class client.BrowserLikeRedirectAgent:
    """
    HTTP agent that follows redirects like a browser.
    """
    def __init__(self, agent, redirectLimit=20):
        """
        Args:
            agent: Underlying HTTP agent
            redirectLimit (int): Maximum number of redirects to follow
        """

class client.CookieAgent:
    """
    HTTP agent that automatically handles cookies.
    """
    def __init__(self, agent, cookieJar):
        """
        Args:
            agent: Underlying HTTP agent
            cookieJar: Cookie jar for storing cookies
        """

class client.ContentDecoderAgent:
    """
    HTTP agent that automatically decodes content (gzip, deflate).
    """
    def __init__(self, agent, decoders=None):
        """
        Args:
            agent: Underlying HTTP agent
            decoders: List of content decoders
        """

class client.ProxyAgent:
    """
    HTTP agent that routes requests through a proxy.
    """
    def __init__(self, endpoint, reactor=None, pool=None):
        """
        Args:
            endpoint: Proxy server endpoint
            reactor: The reactor to use
            pool: Connection pool
        """

class client.FileBodyProducer:
    """
    Body producer for uploading files.
    """
    def __init__(self, inputFile, cooperator=None):
        """
        Args:
            inputFile: File-like object to read from
            cooperator: Task cooperator for yielding control
        """

class client.HTTPConnectionPool:
    """
    Connection pool for HTTP connections.
    """
    def __init__(self, reactor, persistent=True, maxPersistentPerHost=10, 
                 cachedConnectionTimeout=240, retryAutomatically=True):
        """
        Args:
            reactor: The reactor to use
            persistent (bool): Whether to keep connections alive
            maxPersistentPerHost (int): Max connections per host
            cachedConnectionTimeout (int): Connection cache timeout
            retryAutomatically (bool): Retry failed connections
        """

HTTP Client Usage Example:

from twisted.internet import reactor, defer
from twisted.web import client
from twisted.web.http_headers import Headers

@defer.inlineCallbacks
def makeRequest():
    agent = client.Agent(reactor)
    
    # Simple GET request
    response = yield agent.request(
        b'GET',
        b'https://httpbin.org/get',
        Headers({'User-Agent': ['Twisted HTTP Client']})
    )
    
    print(f"Status: {response.code}")
    body = yield client.readBody(response)
    print(f"Body: {body.decode()}")

# Enhanced agent with redirects and cookies
agent = client.Agent(reactor)
redirect_agent = client.BrowserLikeRedirectAgent(agent)
cookie_agent = client.CookieAgent(redirect_agent, cookielib.CookieJar())

makeRequest()
reactor.run()

HTTP Server

HTTP server implementation with resource-based routing, session management, and WSGI support.

class server.Site:
    """
    HTTP server site that serves resources.
    """
    def __init__(self, resource, timeout=60*60*12, logPath=None):
        """
        Args:
            resource: Root resource to serve
            timeout (int): Session timeout in seconds
            logPath: Path for access logs
        """
    
    def getResourceFor(self, request):
        """
        Get the resource for a request.
        
        Args:
            request: HTTP request object
            
        Returns:
            Resource: Resource to handle request
        """

class server.Request:
    """
    HTTP request object.
    
    Key Attributes:
    - method: HTTP method (bytes)
    - uri: Request URI (bytes)  
    - path: Request path (bytes)
    - args: Query parameters (dict)
    - headers: Request headers
    - content: Request body stream
    - cookies: Request cookies (list)
    - session: Session object
    - client: Client address
    """
    method = None
    uri = None
    path = None
    args = None
    headers = None
    content = None
    cookies = None
    
    def getSession(self):
        """Get or create session for this request."""
    
    def write(self, data):
        """
        Write data to response.
        
        Args:
            data (bytes): Data to write
        """
    
    def finish(self):
        """Finish the HTTP response."""
    
    def setHeader(self, name, value):
        """
        Set HTTP response header.
        
        Args:
            name (str): Header name
            value (str): Header value  
        """
    
    def setResponseCode(self, code, message=None):
        """
        Set HTTP response code.
        
        Args:
            code (int): HTTP status code
            message (str): Status message
        """
    
    def redirect(self, url):
        """
        Redirect to another URL.
        
        Args:
            url (bytes): Redirect URL
        """

class server.Session:
    """
    HTTP session for maintaining state across requests.
    
    Attributes:
    - uid: Unique session identifier
    - expireCallbacks: List of expiration callbacks
    """
    uid = None
    
    def expire(self):
        """Expire this session."""
    
    def touch(self):
        """Update session last access time."""

# Server constants
server.NOT_DONE_YET  # Constant indicating async response

HTTP Server Usage Example:

from twisted.web import server, resource
from twisted.internet import reactor, endpoints

class HelloResource(resource.Resource):
    isLeaf = True
    
    def render_GET(self, request):
        request.setHeader(b'content-type', b'text/plain')
        return b'Hello, World!'
    
    def render_POST(self, request):
        data = request.content.read()
        request.setHeader(b'content-type', b'text/plain')
        return f'Received: {data.decode()}'.encode()

root = HelloResource()
site = server.Site(root)

endpoint = endpoints.TCP4ServerEndpoint(reactor, 8080)
endpoint.listen(site)

print("Server listening on port 8080")
reactor.run()

Web Resources

Resource-based system for organizing web applications with hierarchical URL routing and content generation.

class resource.Resource:
    """
    Base class for web resources.
    
    Attributes:
    - isLeaf: Whether this resource has children (bool)
    """
    isLeaf = False
    
    def render(self, request):
        """
        Render this resource for a request.
        
        Args:
            request: HTTP request object
            
        Returns:
            bytes or NOT_DONE_YET: Response data
        """
    
    def render_GET(self, request):
        """Handle GET requests."""
    
    def render_POST(self, request):
        """Handle POST requests."""
    
    def render_PUT(self, request):
        """Handle PUT requests."""
    
    def render_DELETE(self, request):
        """Handle DELETE requests."""
    
    def getChild(self, path, request):
        """
        Get child resource for path segment.
        
        Args:
            path (bytes): Path segment
            request: HTTP request
            
        Returns:
            Resource: Child resource
        """
    
    def putChild(self, path, child):
        """
        Add a child resource.
        
        Args:
            path (bytes): Path segment
            child (Resource): Child resource
        """

class resource.ErrorPage(resource.Resource):
    """
    Resource for generating error pages.
    """
    def __init__(self, status, brief, detail):
        """
        Args:
            status (int): HTTP status code
            brief (str): Brief error description
            detail (str): Detailed error description
        """

class resource.NoResource(resource.ErrorPage):
    """
    404 Not Found resource.
    """
    def __init__(self, message="Sorry. No such resource."):
        """
        Args:
            message (str): 404 error message
        """

class resource.ForbiddenResource(resource.ErrorPage):
    """
    403 Forbidden resource.
    """
    def __init__(self, message="Sorry, resource is forbidden."):
        """
        Args:
            message (str): 403 error message
        """

def resource.getChildForRequest(resource, request):
    """
    Get the appropriate child resource for a request.
    
    Args:
        resource: Parent resource
        request: HTTP request
        
    Returns:
        Resource: Child resource to handle request
    """

Static File Serving

Resources for serving static files and directory listings.

class static.File:
    """
    Resource for serving static files and directories.
    """
    def __init__(self, path, defaultType="text/html", ignoredExts=None, registry=None, allowExt=0):
        """
        Args:
            path (str): File system path to serve
            defaultType (str): Default MIME type
            ignoredExts: File extensions to ignore
            registry: MIME type registry
            allowExt (bool): Allow extension-based MIME detection
        """
    
    def directoryListing(self):
        """
        Get directory listing resource.
        
        Returns:
            Resource: Directory listing resource
        """

class static.DirectoryLister:
    """
    Resource for generating directory listings.
    """
    def __init__(self, pathname, dirs=None, contentTypes=None, contentEncodings=None, defaultType="text/html"):
        """
        Args:
            pathname (str): Directory path
            dirs: Subdirectories to list
            contentTypes: MIME type mappings
            contentEncodings: Content encoding mappings
            defaultType (str): Default MIME type
        """

class static.Data:
    """
    Resource for serving data from memory.
    """
    def __init__(self, data, type):
        """
        Args:
            data (bytes): Data to serve
            type (str): MIME type
        """

WSGI Integration

Support for running WSGI applications within Twisted's web server.

class wsgi.WSGIResource:
    """
    Resource for running WSGI applications.
    """
    def __init__(self, reactor, threadpool, application):
        """
        Args:
            reactor: The reactor to use
            threadpool: Thread pool for WSGI app
            application: WSGI application callable
        """

WSGI Usage Example:

from twisted.web import wsgi, server
from twisted.python import threadpool
from twisted.internet import reactor, endpoints

# Simple WSGI application
def simple_app(environ, start_response):
    status = '200 OK'
    headers = [('Content-type', 'text/plain')]
    start_response(status, headers)
    return [b'Hello from WSGI!']

# Create WSGI resource
pool = threadpool.ThreadPool()
pool.start()
wsgi_resource = wsgi.WSGIResource(reactor, pool, simple_app)

# Serve it
site = server.Site(wsgi_resource)
endpoint = endpoints.TCP4ServerEndpoint(reactor, 8080)
endpoint.listen(site)

HTTP Headers

Utilities for working with HTTP headers.

class http_headers.Headers:
    """
    HTTP headers container.
    """
    def __init__(self, rawHeaders=None):
        """
        Args:
            rawHeaders: Dict of header name to list of values
        """
    
    def hasHeader(self, name):
        """
        Check if header exists.
        
        Args:
            name (str): Header name
            
        Returns:
            bool: True if header exists
        """
    
    def getHeader(self, name, default=None):
        """
        Get header value.
        
        Args:
            name (str): Header name
            default: Default value if not found
            
        Returns:
            bytes or default: Header value
        """
    
    def setHeader(self, name, value):
        """
        Set header value.
        
        Args:
            name (str): Header name
            value: Header value (str or list)
        """
    
    def addRawHeader(self, name, value):
        """
        Add raw header value.
        
        Args:
            name (bytes): Header name
            value (bytes): Header value
        """

Template System

Basic template system for generating dynamic HTML content.

class template.Element:
    """
    Base class for template elements.
    """
    def __init__(self, loader, tag=None):
        """
        Args:
            loader: Template loader
            tag: Root template tag
        """

class template.XMLFile:
    """
    Load template from XML file.
    """
    def __init__(self, path):
        """
        Args:
            path (str): Path to XML template file
        """

class template.XMLString:
    """
    Load template from XML string.
    """
    def __init__(self, template):
        """
        Args:
            template (str): XML template string
        """

def template.renderer(func):
    """
    Decorator for template renderer methods.
    
    Args:
        func: Renderer function
        
    Returns:
        Decorated renderer function
    """

def template.flatten(element):
    """
    Flatten template element to bytes.
    
    Args:
        element: Template element
        
    Returns:
        Deferred[bytes]: Rendered template
    """

Web Utilities

Helper functions and utilities for web development.

def util.redirectTo(URL, request):
    """
    Generate redirect response.
    
    Args:
        URL (bytes): Redirect URL
        request: HTTP request object
        
    Returns:
        bytes: Redirect response data
    """

class util.Redirect:
    """
    Resource that redirects to another URL.
    """
    def __init__(self, url):
        """
        Args:
            url (bytes): Redirect URL
        """

class util.ChildRedirector:
    """
    Resource that redirects based on path.
    """
    def __init__(self, url):
        """
        Args:
            url (bytes): Base redirect URL
        """

Interface Types

Key interfaces and types used in Twisted's HTTP implementation.

# From twisted.web.iweb
class IBodyProducer:
    """
    Interface for request body producers.
    
    Methods:
    - startProducing(consumer): Start producing data
    - stopProducing(): Stop producing data
    - pauseProducing(): Pause data production  
    - resumeProducing(): Resume data production
    """
    def startProducing(consumer): ...
    def stopProducing(): ...
    def pauseProducing(): ...
    def resumeProducing(): ...

class IResponse:
    """
    Interface for HTTP responses.
    
    Attributes:
    - version: HTTP version tuple
    - code: Status code
    - phrase: Status phrase
    - headers: Response headers
    - length: Content length
    """
    version = None
    code = None
    phrase = None  
    headers = None
    length = None

# From twisted.web.http_headers
class Headers:
    """
    HTTP header collection.
    
    Methods:
    - addRawHeader(name, value): Add header
    - setRawHeaders(name, values): Set header values
    - getRawHeaders(name): Get header values
    - hasHeader(name): Check if header exists
    """
    def __init__(self, rawHeaders=None): ...
    def addRawHeader(self, name, value): ...
    def setRawHeaders(self, name, values): ...
    def getRawHeaders(self, name, default=None): ...
    def hasHeader(self, name): ...

# Constants
NOT_DONE_YET = object()  # Indicates async response
UNKNOWN_LENGTH = -1      # Unknown content length

Required Imports

Complete import statements for HTTP functionality:

# Core HTTP functionality
from twisted.web import client, server, resource, static, http
from twisted.web.http_headers import Headers
from twisted.web.iweb import IBodyProducer, IResponse
from twisted.internet import defer, reactor, endpoints

# Common import patterns
from twisted.web.client import (
    Agent, Response, BrowserLikeRedirectAgent, CookieAgent, 
    ContentDecoderAgent, ProxyAgent, FileBodyProducer, readBody
)
from twisted.web.server import Site, Request, Session  
from twisted.web.resource import Resource, ErrorPage, NoResource
from twisted.web.static import File

Install with Tessl CLI