tessl/pypi-aiohttp

Async http client/server framework (asyncio)

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Web Server

Name: tessl/pypi-aiohttp
Author: tessl

Full-featured asynchronous web server framework for building modern web applications and APIs. Supports applications, routing, middleware, static file serving, and comprehensive request/response handling with extensive customization options.

Capabilities

Application Management

The core application container that manages routes, middleware, and application lifecycle.

class Application:
    def __init__(
        self,
        *,
        logger=None,
        router=None,
        middlewares=None,
        handler_args=None,
        client_max_size=1024**2,
        debug=False,
        **kwargs
    ):
        """
        Create web application.
        
        Parameters:
        - logger: Application logger
        - router: URL router instance
        - middlewares: List of middleware functions
        - handler_args: Arguments for request handlers
        - client_max_size (int): Maximum client payload size
        - debug (bool): Enable debug mode
        """
    
    @property
    def router(self):
        """Application URL router."""
    
    def add_routes(self, routes):
        """Add multiple routes to application."""
    
    def cleanup_ctx(self, coro):
        """Add cleanup context manager."""
    
    def on_startup(self, coro):
        """Add startup hook."""
    
    def on_shutdown(self, coro):
        """Add shutdown hook."""
    
    def on_cleanup(self, coro):
        """Add cleanup hook."""
    
    def __getitem__(self, key):
        """Get application data."""
    
    def __setitem__(self, key, value):
        """Set application data."""

class AppKey:
    def __init__(self, name, t):
        """
        Type-safe application key.
        
        Parameters:
        - name (str): Key name
        - t (type): Expected value type
        """

Request and Response Handling

Objects representing HTTP requests and responses with comprehensive data access and manipulation methods.

class Request:
    @property
    def method(self):
        """HTTP request method."""
    
    @property
    def url(self):
        """Request URL."""
    
    @property
    def path(self):
        """Request path."""
    
    @property
    def query(self):
        """Query parameters."""
    
    @property
    def headers(self):
        """Request headers."""
    
    @property
    def cookies(self):
        """Request cookies."""
    
    @property
    def content_type(self):
        """Request content type."""
    
    @property
    def content_length(self):
        """Request content length."""
    
    async def read(self):
        """Read request body as bytes."""
    
    async def text(self):
        """Read request body as text."""
    
    async def json(self):
        """Parse request body as JSON."""
    
    async def post(self):
        """Parse form data from request body."""
    
    async def multipart(self):
        """Parse multipart data from request body."""

class BaseRequest:
    """Base class for HTTP requests."""

class Response:
    def __init__(
        self,
        *,
        body=None, 
        text=None,
        status=200,
        reason=None,
        headers=None,
        content_type=None,
        charset=None,
        **kwargs
    ):
        """
        Create HTTP response.
        
        Parameters:
        - body (bytes): Response body as bytes
        - text (str): Response body as text
        - status (int): HTTP status code
        - reason (str): HTTP reason phrase
        - headers (dict): Response headers
        - content_type (str): Content type
        - charset (str): Text encoding
        """
    
    @property
    def status(self):
        """HTTP status code."""
    
    @property
    def headers(self):
        """Response headers."""
    
    @property
    def content_type(self):
        """Response content type."""

class StreamResponse:
    def __init__(
        self,
        *,
        status=200,
        reason=None,
        headers=None,
        **kwargs
    ):
        """Streaming HTTP response."""
    
    async def prepare(self, request):
        """Prepare response for streaming."""
    
    async def write(self, data):
        """Write data to response stream."""
    
    async def write_eof(self):
        """Signal end of response."""

def json_response(
    data=None,
    *,
    text=None,
    body=None,
    status=200,
    reason=None,
    headers=None,
    content_type='application/json',
    dumps=None,
    **kwargs
):
    """
    Create JSON response.
    
    Parameters:
    - data: Object to serialize as JSON
    - text (str): JSON text
    - body (bytes): JSON body
    - status (int): HTTP status code
    - reason (str): HTTP reason phrase  
    - headers (dict): Response headers
    - content_type (str): Content type
    - dumps: JSON serialization function
    
    Returns:
    Response: JSON response object
    """

class FileResponse(StreamResponse):
    def __init__(
        self,
        path,
        *,
        chunk_size=256*1024,
        status=200,
        reason=None,
        headers=None,
        **kwargs
    ):
        """
        File-based HTTP response.
        
        Parameters:
        - path: File path
        - chunk_size (int): Streaming chunk size
        - status (int): HTTP status code
        - reason (str): HTTP reason phrase
        - headers (dict): Response headers
        """

URL Routing and Dispatching

Comprehensive URL routing system with support for static routes, dynamic parameters, and resource-based organization.

class UrlDispatcher:
    def add_get(self, path, handler, **kwargs):
        """Add GET route."""
    
    def add_post(self, path, handler, **kwargs):
        """Add POST route."""
    
    def add_put(self, path, handler, **kwargs):
        """Add PUT route."""
    
    def add_patch(self, path, handler, **kwargs):
        """Add PATCH route."""
    
    def add_delete(self, path, handler, **kwargs):
        """Add DELETE route."""
    
    def add_head(self, path, handler, **kwargs):
        """Add HEAD route."""
    
    def add_options(self, path, handler, **kwargs):
        """Add OPTIONS route."""
    
    def add_route(self, method, path, handler, **kwargs):
        """Add route for specific HTTP method."""
    
    def add_static(self, prefix, path, **kwargs):
        """Add static file serving route."""
    
    def add_resource(self, path, **kwargs):
        """Add URL resource."""

class AbstractResource:
    """Base class for URL resources."""

class Resource(AbstractResource):
    """URL resource with route collection."""

class DynamicResource(Resource):
    """Dynamic URL resource with path variables."""

class StaticResource(Resource):
    """Static file serving resource."""

class View:
    """Class-based view for handling requests."""
    
    async def get(self):
        """Handle GET request."""
    
    async def post(self):
        """Handle POST request."""
    
    async def put(self):
        """Handle PUT request."""
    
    async def patch(self):
        """Handle PATCH request."""
    
    async def delete(self):
        """Handle DELETE request."""

Route Definition Decorators

Decorator-based route definition system for clean and organized route registration.

class RouteTableDef:
    def __init__(self):
        """Collection of route definitions."""
    
    def get(self, path, **kwargs):
        """GET route decorator."""
    
    def post(self, path, **kwargs):
        """POST route decorator."""
    
    def put(self, path, **kwargs):
        """PUT route decorator."""
    
    def patch(self, path, **kwargs):
        """PATCH route decorator."""
    
    def delete(self, path, **kwargs):
        """DELETE route decorator."""
    
    def head(self, path, **kwargs):
        """HEAD route decorator."""
    
    def options(self, path, **kwargs):
        """OPTIONS route decorator."""
    
    def route(self, method, path, **kwargs):
        """Generic route decorator."""
    
    def view(self, path, **kwargs):
        """Class-based view decorator."""
    
    def static(self, prefix, path, **kwargs):
        """Static file route decorator."""

def get(path, **kwargs):
    """Standalone GET route decorator."""

def post(path, **kwargs):
    """Standalone POST route decorator."""

def put(path, **kwargs):
    """Standalone PUT route decorator."""

def patch(path, **kwargs):
    """Standalone PATCH route decorator."""

def delete(path, **kwargs):
    """Standalone DELETE route decorator."""

def head(path, **kwargs):
    """Standalone HEAD route decorator."""

def options(path, **kwargs):
    """Standalone OPTIONS route decorator."""

def route(method, path, **kwargs):
    """Standalone generic route decorator."""

def view(path, **kwargs):
    """Standalone class-based view decorator."""

def static(prefix, path, **kwargs):
    """Standalone static file route decorator."""

Middleware System

Extensible middleware system for request/response processing, authentication, and cross-cutting concerns.

def middleware(coro):
    """
    Decorator for creating middleware functions.
    
    Parameters:
    - coro: Async function that processes requests
    
    Returns:
    Middleware function
    """

def normalize_path_middleware(
    *,
    append_slash=True,
    remove_slash=False,
    merge_slashes=True,
    redirect_class=None
):
    """
    Path normalization middleware.
    
    Parameters:
    - append_slash (bool): Append trailing slash
    - remove_slash (bool): Remove trailing slash
    - merge_slashes (bool): Merge consecutive slashes
    - redirect_class: HTTP redirect class
    
    Returns:
    Middleware function
    """

Application Runner and Sites

Application lifecycle management and server binding for different transport types.

class AppRunner:
    def __init__(self, app, **kwargs):
        """
        Application runner.
        
        Parameters:
        - app: Web application instance
        """
    
    async def setup(self):
        """Setup application runner."""
    
    async def cleanup(self):
        """Cleanup application runner."""

class BaseRunner:
    """Base class for application runners."""

class ServerRunner(BaseRunner):
    def __init__(self, web_server, **kwargs):
        """Runner for raw request handlers."""

class BaseSite:
    """Base class for server sites."""

class TCPSite(BaseSite):
    def __init__(
        self,
        runner,
        host=None,
        port=None,
        *,
        ssl_context=None,
        backlog=128,
        reuse_address=None,
        reuse_port=None,
        **kwargs
    ):
        """
        TCP server site.
        
        Parameters:
        - runner: Application runner
        - host (str): Bind host
        - port (int): Bind port
        - ssl_context: SSL context
        - backlog (int): Socket backlog
        - reuse_address (bool): Reuse address
        - reuse_port (bool): Reuse port
        """
    
    async def start(self):
        """Start server site."""
    
    async def stop(self):
        """Stop server site."""

class UnixSite(BaseSite):
    def __init__(self, runner, path, **kwargs):
        """Unix domain socket site."""

class NamedPipeSite(BaseSite):
    def __init__(self, runner, path, **kwargs):
        """Named pipe site (Windows)."""

def run_app(
    app,
    *,
    host=None,
    port=None,
    path=None,
    sock=None,
    shutdown_timeout=60.0,
    keepalive_timeout=75.0,
    ssl_context=None,
    print=print,
    backlog=128,
    access_log_class=None,
    access_log_format=None,
    access_log=None,
    handle_signals=True,
    reuse_address=None,
    reuse_port=None,
    handler_cancellation=False,
    loop=None
):
    """
    Run web application.
    
    Parameters:
    - app: Web application or coroutine returning application
    - host (str or list): Server host(s)
    - port (int): Server port
    - path (str or list): Unix socket path(s)
    - sock: Raw socket or list of sockets
    - shutdown_timeout (float): Shutdown timeout
    - keepalive_timeout (float): Keep-alive timeout
    - ssl_context: SSL context
    - print: Print function for startup messages
    - backlog (int): Socket backlog
    - access_log_class: Access log class
    - access_log_format (str): Access log format
    - access_log: Access logger instance
    - handle_signals (bool): Handle OS signals
    - reuse_address (bool): Reuse address option
    - reuse_port (bool): Reuse port option
    - handler_cancellation (bool): Enable handler cancellation
    - loop: Event loop
    """

HTTP Status Code Exceptions

Comprehensive HTTP status code exception classes for proper HTTP response handling.

class HTTPException(Exception):
    """Base HTTP exception."""
    
    @property
    def status_code(self):
        """HTTP status code."""

# Success responses (2xx)
class HTTPOk(HTTPException):
    """200 OK response."""

class HTTPCreated(HTTPException):
    """201 Created response."""

class HTTPNoContent(HTTPException):
    """204 No Content response."""

# Redirection responses (3xx)
class HTTPMovedPermanently(HTTPException):
    """301 Moved Permanently response."""

class HTTPFound(HTTPException):
    """302 Found response."""

class HTTPNotModified(HTTPException):
    """304 Not Modified response."""

# Client error responses (4xx)
class HTTPBadRequest(HTTPException):
    """400 Bad Request response."""

class HTTPUnauthorized(HTTPException):
    """401 Unauthorized response."""

class HTTPForbidden(HTTPException):
    """403 Forbidden response."""

class HTTPNotFound(HTTPException):
    """404 Not Found response."""

class HTTPMethodNotAllowed(HTTPException):
    """405 Method Not Allowed response."""

class HTTPConflict(HTTPException):
    """409 Conflict response."""

# Server error responses (5xx)
class HTTPInternalServerError(HTTPException):
    """500 Internal Server Error response."""

class HTTPNotImplemented(HTTPException):
    """501 Not Implemented response."""

class HTTPBadGateway(HTTPException):
    """502 Bad Gateway response."""

class HTTPServiceUnavailable(HTTPException):
    """503 Service Unavailable response."""

Usage Examples

Basic Web Application

from aiohttp import web

async def hello_handler(request):
    name = request.match_info.get('name', 'World')
    return web.Response(text=f"Hello, {name}!")

async def json_handler(request):
    data = {'message': 'Hello, JSON!', 'status': 'success'}
    return web.json_response(data)

# Create application
app = web.Application()

# Add routes
app.router.add_get('/', hello_handler)
app.router.add_get('/hello/{name}', hello_handler)
app.router.add_get('/json', json_handler)

# Run application
if __name__ == '__main__':
    web.run_app(app, host='localhost', port=8080)

Route Table Definitions

from aiohttp import web

routes = web.RouteTableDef()

@routes.get('/')
async def index(request):
    return web.Response(text="Welcome to the API")

@routes.get('/users/{user_id}')
async def get_user(request):
    user_id = request.match_info['user_id']
    return web.json_response({'user_id': user_id})

@routes.post('/users')
async def create_user(request):
    data = await request.json()
    # Process user creation
    return web.json_response({'created': True}, status=201)

# Create application and add routes
app = web.Application()
app.add_routes(routes)

web.run_app(app, port=8080)

Middleware Usage

from aiohttp import web
import time

@web.middleware
async def timing_middleware(request, handler):
    start_time = time.time()
    response = await handler(request)
    process_time = time.time() - start_time
    response.headers['X-Process-Time'] = str(process_time)
    return response

@web.middleware
async def auth_middleware(request, handler):
    auth_header = request.headers.get('Authorization')
    if not auth_header:
        raise web.HTTPUnauthorized()
    
    # Validate authentication
    return await handler(request)

# Create application with middleware
app = web.Application(middlewares=[timing_middleware, auth_middleware])

async def protected_handler(request):
    return web.json_response({'message': 'Access granted'})

app.router.add_get('/protected', protected_handler)

web.run_app(app)

Class-Based Views

from aiohttp import web

class UserView(web.View):
    async def get(self):
        user_id = self.request.match_info['user_id']
        # Fetch user data
        return web.json_response({'user_id': user_id})
    
    async def post(self):
        data = await self.request.json()
        # Create user
        return web.json_response({'created': True}, status=201)
    
    async def put(self):
        user_id = self.request.match_info['user_id']
        data = await self.request.json()
        # Update user
        return web.json_response({'updated': True})
    
    async def delete(self):
        user_id = self.request.match_info['user_id']
        # Delete user
        return web.Response(status=204)

# Add class-based view to router
app = web.Application()
app.router.add_view('/users/{user_id}', UserView)

web.run_app(app)

Install with Tessl CLI