tessl/pypi-hypercorn

A high-performance ASGI and WSGI web server implementation that provides comprehensive support for modern web protocols including HTTP/1, HTTP/2, WebSockets, and experimental HTTP/3

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Logging

Name: tessl/pypi-hypercorn
Author: tessl

Comprehensive logging framework with access log formatting, structured logging, and StatsD metrics integration for monitoring and observability. Hypercorn provides flexible logging capabilities suitable for both development and production environments.

Capabilities

Logger Class

Main logging class that provides async logging methods and access log functionality with configurable formatting and output destinations.

class Logger:
    """
    Main logging class for Hypercorn.
    
    Provides async logging methods for different log levels and
    specialized access logging with configurable formatting.
    Supports multiple output destinations and structured logging.
    """
    
    def __init__(self, config: Config):
        """
        Initialize logger with configuration.
        
        Args:
            config: Config object containing logging settings
                   including log files, levels, and formatting
        """
        
    @property
    def handlers(self) -> list:
        """
        List of active log handlers.
        
        Returns:
            List of logging handlers (file, stream, etc.)
        """
        
    async def critical(self, message: str, *args, **kwargs):
        """
        Log critical error message.
        
        Args:
            message: Log message string (supports % formatting)
            *args: Positional arguments for message formatting
            **kwargs: Keyword arguments for message formatting
            
        Critical messages indicate severe errors that may cause
        the server to terminate or become unusable.
        """
        
    async def error(self, message: str, *args, **kwargs):
        """
        Log error message.
        
        Args:
            message: Log message string (supports % formatting)
            *args: Positional arguments for message formatting
            **kwargs: Keyword arguments for message formatting
            
        Error messages indicate problems that don't prevent
        continued operation but need attention.
        """
        
    async def warning(self, message: str, *args, **kwargs):
        """
        Log warning message.
        
        Args:
            message: Log message string (supports % formatting)
            *args: Positional arguments for message formatting
            **kwargs: Keyword arguments for message formatting
            
        Warning messages indicate potential issues or
        unexpected conditions that don't prevent operation.
        """
        
    async def info(self, message: str, *args, **kwargs):
        """
        Log informational message.
        
        Args:
            message: Log message string (supports % formatting)
            *args: Positional arguments for message formatting
            **kwargs: Keyword arguments for message formatting
            
        Info messages provide general operational information
        about server status and request processing.
        """
        
    async def debug(self, message: str, *args, **kwargs):
        """
        Log debug message.
        
        Args:
            message: Log message string (supports % formatting)
            *args: Positional arguments for message formatting
            **kwargs: Keyword arguments for message formatting
            
        Debug messages provide detailed information for
        troubleshooting and development purposes.
        """
        
    async def exception(self, message: str, *args, **kwargs):
        """
        Log exception with traceback.
        
        Args:
            message: Log message string (supports % formatting)
            *args: Positional arguments for message formatting
            **kwargs: Keyword arguments for message formatting
            
        Exception messages include full stack traces and are
        typically called from exception handlers.
        """
        
    async def log(self, level: int, message: str, *args, **kwargs):
        """
        Log message at specified level.
        
        Args:
            level: Logging level (from logging module constants)
            message: Log message string (supports % formatting)
            *args: Positional arguments for message formatting
            **kwargs: Keyword arguments for message formatting
            
        Generic logging method that can log at any level.
        """
        
    async def access(self, request, response, request_time: float):
        """
        Log access entry for HTTP request.
        
        Args:
            request: Request object containing request information
            response: Response object containing response information
            request_time: Time taken to process request in seconds
            
        Creates formatted access log entries using the configured
        access log format string and available log atoms.
        """
        
    def atoms(self, request, response, request_time: float) -> dict[str, str]:
        """
        Get access log atoms for formatting.
        
        Args:
            request: Request object containing request information
            response: Response object containing response information  
            request_time: Time taken to process request in seconds
            
        Returns:
            Dictionary of format atoms for access log formatting
        """

Access Log Atoms

Dictionary-like class that provides access log formatting variables for creating custom access log formats.

class AccessLogAtoms:
    """
    Dictionary-like class for access log formatting.
    
    Provides access to various request and response attributes
    for use in access log format strings. Similar to Apache's
    log format variables.
    """
    
    def __init__(self, request, response, request_time: float):
        """
        Initialize access log atoms.
        
        Args:
            request: HTTP request object
            response: HTTP response object  
            request_time: Request processing time in seconds
        """
        
    # Available atoms (accessed via dictionary interface):
    # h - Remote hostname/IP address
    # l - Remote logname (always '-')
    # u - Remote user (from authentication)
    # t - Time of request in common log format
    # r - First line of request ("GET /path HTTP/1.1")
    # s - Response status code
    # b - Response body size in bytes
    # f - Referer header
    # a - User agent header
    # T - Request processing time in seconds
    # D - Request processing time in microseconds
    # L - Request processing time in decimal seconds
    # p - Server port
    # P - Process ID
    # {header}i - Request header value
    # {header}o - Response header value
    # {variable}e - Environment variable value

StatsD Logger

Extended logger class that integrates with StatsD metrics systems for monitoring and observability, providing both logging and metrics collection.

class StatsdLogger(Logger):
    """
    Logger with StatsD metrics integration.
    
    Extends the base Logger class with StatsD metrics capabilities,
    enabling both traditional logging and metrics collection for
    monitoring, alerting, and performance analysis.
    """
    
    def __init__(self, config: Config, statsd_host: str = "localhost", statsd_port: int = 8125):
        """
        Initialize StatsD logger.
        
        Args:
            config: Config object with logging settings
            statsd_host: StatsD server hostname
            statsd_port: StatsD server port
        """
        
    def increment(self, name: str, value: int = 1, tags: dict | None = None):
        """
        Increment a counter metric.
        
        Args:
            name: Metric name (e.g., "requests.total")
            value: Increment value (default: 1)
            tags: Optional tags/labels dictionary
            
        Increments a counter metric by the specified value.
        Used for counting events like requests, errors, etc.
        """
        
    def decrement(self, name: str, value: int = 1, tags: dict | None = None):
        """
        Decrement a counter metric.
        
        Args:
            name: Metric name (e.g., "connections.active")
            value: Decrement value (default: 1)
            tags: Optional tags/labels dictionary
            
        Decrements a counter metric by the specified value.
        """
        
    def histogram(self, name: str, value: float, tags: dict | None = None):
        """
        Record a histogram/timing metric.
        
        Args:
            name: Metric name (e.g., "request.duration")
            value: Metric value (typically timing in seconds)
            tags: Optional tags/labels dictionary
            
        Records a timing or distribution metric value.
        Used for request durations, response sizes, etc.
        """
        
    def gauge(self, name: str, value: float, tags: dict | None = None):
        """
        Set a gauge metric value.
        
        Args:
            name: Metric name (e.g., "memory.usage")
            value: Current metric value
            tags: Optional tags/labels dictionary
            
        Sets a gauge metric to the specified value.
        Used for current state metrics like memory usage,
        active connections, queue sizes, etc.
        """

Usage Examples

Basic Logger Usage

from hypercorn.config import Config
from hypercorn.logging import Logger

# Create logger with configuration
config = Config()
config.errorlog = "/var/log/hypercorn-error.log"
config.accesslog = "/var/log/hypercorn-access.log"
config.access_log_format = '%(h)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s"'

logger = Logger(config)

# Use async logging methods
async def handle_request():
    await logger.info("Processing request")
    
    try:
        # Process request
        result = await process_request()
        await logger.debug("Request processed successfully")
        return result
    except Exception as e:
        await logger.error("Request processing failed: %s", str(e))
        await logger.exception("Full exception details")
        raise

Access Logging

# Access logging is typically handled automatically by Hypercorn
# but can be used manually:

async def log_request_completion(request, response, start_time):
    request_time = time.time() - start_time
    await logger.access(request, response, request_time)

# Custom access log format in config
config.access_log_format = (
    '%(h)s - %(u)s [%(t)s] "%(r)s" %(s)s %(b)s '
    '"%(f)s" "%(a)s" %(T)s'
)

StatsD Integration

from hypercorn.logging import StatsdLogger

# Create StatsD logger
config = Config()
statsd_logger = StatsdLogger(
    config, 
    statsd_host="metrics.example.com",
    statsd_port=8125
)

# Use metrics alongside logging
async def handle_request():
    # Increment request counter
    statsd_logger.increment("requests.total", tags={"method": "GET"})
    
    start_time = time.time()
    try:
        # Process request
        result = await process_request()
        
        # Record success metrics
        statsd_logger.increment("requests.success")
        statsd_logger.histogram("request.duration", time.time() - start_time)
        
        await statsd_logger.info("Request completed successfully")
        return result
        
    except Exception as e:
        # Record error metrics
        statsd_logger.increment("requests.error", tags={"error_type": type(e).__name__})
        await statsd_logger.error("Request failed: %s", str(e))
        raise

Custom Access Log Format

# Configure custom access log format
config = Config()

# Apache Common Log Format
config.access_log_format = '%(h)s - %(u)s [%(t)s] "%(r)s" %(s)s %(b)s'

# Apache Combined Log Format  
config.access_log_format = (
    '%(h)s - %(u)s [%(t)s] "%(r)s" %(s)s %(b)s '
    '"%(f)s" "%(a)s"'
)

# Custom format with timing
config.access_log_format = (
    '%(h)s "%(r)s" %(s)s %(b)s %(T)s '
    'pid=%(P)s port=%(p)s'
)

# JSON format for structured logging
config.access_log_format = (
    '{"remote_addr": "%(h)s", "request": "%(r)s", '
    '"status": %(s)s, "bytes": %(b)s, "duration": %(T)s}'
)

Production Logging Setup

from hypercorn.config import Config
from hypercorn.logging import StatsdLogger

# Production logging configuration
config = Config()

# Log files with rotation (handled by external tools)
config.errorlog = "/var/log/hypercorn/error.log"
config.accesslog = "/var/log/hypercorn/access.log"

# Structured access logs
config.access_log_format = (
    '{"timestamp": "%(t)s", "client_ip": "%(h)s", '
    '"method": "%(m)s", "path": "%(U)s", "status": %(s)s, '
    '"bytes": %(b)s, "duration": %(T)s, "user_agent": "%(a)s"}'
)

# Create StatsD logger for metrics
logger = StatsdLogger(config, statsd_host="metrics.internal")

# Set up log levels
import logging
logging.getLogger("hypercorn.error").setLevel(logging.INFO)
logging.getLogger("hypercorn.access").setLevel(logging.INFO)

Development Logging

# Development configuration with debug output
config = Config()
config.debug = True

# Log to stdout/stderr for development
config.errorlog = "-"  # stderr
config.accesslog = "-"  # stdout

# Simple access log format
config.access_log_format = '%(h)s "%(r)s" %(s)s %(T)s'

logger = Logger(config)

# Enable debug logging
import logging
logging.getLogger("hypercorn").setLevel(logging.DEBUG)

Install with Tessl CLI