Utilities

def load_application(path: str, wsgi_max_body_size: int):
    """
    Load application from module path.
    
    Dynamically imports and loads an ASGI or WSGI application from
    a module path specification. Handles both module and attribute
    loading with proper error handling.
    
    Args:
        path: Module path in format "module:attribute" or "module"
              Examples: "myapp:app", "myapp.wsgi:application"
        wsgi_max_body_size: Maximum body size for WSGI applications
        
    Returns:
        Loaded application object (ASGI or WSGI)
        
    Raises:
        ImportError: If module cannot be imported
        AttributeError: If attribute doesn't exist in module
        NoAppError: If loaded object is not a valid application
    """

def wrap_app(app, wsgi_max_body_size: int, mode: str | None = None):
    """
    Wrap application in appropriate wrapper.
    
    Automatically detects application type (ASGI/WSGI) and wraps
    it with the appropriate wrapper class for use with Hypercorn.
    
    Args:
        app: Application object to wrap
        wsgi_max_body_size: Maximum body size for WSGI applications
        mode: Optional mode specification ("asgi" or "wsgi")
              If None, auto-detects based on application signature
              
    Returns:
        Wrapped application (ASGIWrapper or WSGIWrapper instance)
        
    Raises:
        ValueError: If mode is specified but doesn't match application
        TypeError: If application is neither ASGI nor WSGI compatible
    """

def is_asgi(app) -> bool:
    """
    Check if application is ASGI-compatible.
    
    Examines application signature to determine if it follows
    the ASGI interface specification.
    
    Args:
        app: Application object to check
        
    Returns:
        True if application is ASGI-compatible, False otherwise
        
    Checks for:
        - Callable with 3 parameters (scope, receive, send)
        - Async function or method
        - Proper ASGI signature patterns
    """

def files_to_watch() -> dict[Path, float]:
    """
    Get list of files to watch for changes.
    
    Discovers Python modules and other files that should be
    monitored for changes to trigger server reloads during
    development.
    
    Returns:
        Dictionary mapping file paths to modification times
        
    Includes:
        - All imported Python modules
        - Configuration files
        - Template files (if applicable)
        - Static files (if applicable)
    """

def check_for_updates(files: dict[Path, float]) -> bool:
    """
    Check if watched files have been modified.
    
    Examines file modification times to determine if any
    of the watched files have changed since last check.
    
    Args:
        files: Dictionary mapping file paths to last known modification times
        
    Returns:
        True if any files have been modified, False otherwise
        
    Used by auto-reload functionality to trigger server
    restarts when source code changes during development.
    """

def write_pid_file(pid_path: str):
    """
    Write process PID to file.
    
    Creates a PID file containing the current process ID,
    used for process management and monitoring.
    
    Args:
        pid_path: Path where PID file should be written
        
    Raises:
        IOError: If PID file cannot be written
        PermissionError: If insufficient permissions for file location
        
    The PID file is typically used by process managers,
    monitoring systems, and shutdown scripts.
    """

def parse_socket_addr(family: int, address: tuple):
    """
    Parse socket address based on family.
    
    Parses socket addresses for different address families,
    handling IPv4, IPv6, and Unix socket addresses.
    
    Args:
        family: Socket family (socket.AF_INET, socket.AF_INET6, socket.AF_UNIX)
        address: Address tuple from socket operations
        
    Returns:
        Parsed address information (varies by family)
        
    Address formats:
        - AF_INET: (host, port)
        - AF_INET6: (host, port, flowinfo, scopeid)  
        - AF_UNIX: path string
    """

def repr_socket_addr(family: int, address: tuple) -> str:
    """
    Create string representation of socket address.
    
    Formats socket addresses as human-readable strings
    for logging and display purposes.
    
    Args:
        family: Socket family constant
        address: Address tuple
        
    Returns:
        String representation of address
        
    Examples:
        - IPv4: "192.168.1.100:8000"
        - IPv6: "[::1]:8000"
        - Unix: "/tmp/socket"
    """

def build_and_validate_headers(headers) -> list[tuple[bytes, bytes]]:
    """
    Build and validate HTTP headers.
    
    Processes and validates HTTP headers, ensuring proper
    formatting and compliance with HTTP specifications.
    
    Args:
        headers: Headers in various formats (list, dict, etc.)
        
    Returns:
        List of (name, value) byte tuples
        
    Raises:
        ValueError: If headers are malformed
        TypeError: If headers are wrong type
        
    Performs validation:
        - Header name format and characters
        - Header value format and encoding
        - Prohibited headers (e.g., connection-specific)
    """

def filter_pseudo_headers(headers: list[tuple[bytes, bytes]]) -> list[tuple[bytes, bytes]]:
    """
    Filter out HTTP/2 pseudo-headers.
    
    Removes HTTP/2 pseudo-headers (starting with ':') from
    header list, typically when converting from HTTP/2 to HTTP/1.
    
    Args:
        headers: List of (name, value) header tuples
        
    Returns:
        Filtered headers list without pseudo-headers
        
    HTTP/2 pseudo-headers include:
        - :method, :path, :scheme, :authority (request)
        - :status (response)
    """

def suppress_body(method: str, status_code: int) -> bool:
    """
    Check if response body should be suppressed.
    
    Determines whether an HTTP response should include a body
    based on the request method and response status code.
    
    Args:
        method: HTTP request method (GET, HEAD, POST, etc.)
        status_code: HTTP response status code
        
    Returns:
        True if body should be suppressed, False otherwise
        
    Bodies are suppressed for:
        - HEAD requests (always)
        - 1xx informational responses
        - 204 No Content responses
        - 304 Not Modified responses
    """

def valid_server_name(config: Config, request) -> bool:
    """
    Validate server name against configuration.
    
    Checks if the server name from request matches the
    configured valid server names in the server configuration.
    
    Args:
        config: Server configuration object
        request: HTTP request object containing Host header
        
    Returns:
        True if server name is valid, False otherwise
        
    Validation includes:
        - Exact hostname matches
        - Wildcard pattern matching
        - Port number handling
        - Default server name fallback
    """

async def raise_shutdown(shutdown_event: Callable[..., Awaitable]) -> None:
    """
    Raise shutdown signal by awaiting the shutdown event.
    
    Waits for the shutdown event to complete and then raises
    ShutdownError to trigger server shutdown sequence.
    
    Args:
        shutdown_event: Async callable that when awaited triggers shutdown
        
    Raises:
        ShutdownError: Always raised after shutdown event completes
    """

async def check_multiprocess_shutdown_event(
    shutdown_event: EventType, 
    sleep: Callable[[float], Awaitable[Any]]
) -> None:
    """
    Check for multiprocess shutdown events.
    
    Periodically checks for shutdown events in multiprocess
    environments, using the provided sleep function for timing.
    
    Args:
        shutdown_event: Multiprocess event object to check
        sleep: Async sleep function for periodic checking
        
    Used in multiprocess worker implementations to detect
    when the master process signals shutdown.
    """

class NoAppError(Exception):
    """
    Raised when application cannot be loaded.
    
    This exception occurs when the application loading
    process fails due to import errors, missing attributes,
    or invalid application objects.
    """

class ShutdownError(Exception):
    """
    Raised during server shutdown process.
    
    This exception indicates errors that occur during
    the server shutdown sequence, such as timeout issues
    or resource cleanup failures.
    """

class LifespanTimeoutError(Exception):
    """
    Raised when ASGI lifespan events timeout.
    
    This exception occurs when ASGI application lifespan
    startup or shutdown events take longer than the
    configured timeout period.
    """
    
    def __init__(self, stage: str) -> None:
        """
        Initialize lifespan timeout error.
        
        Args:
            stage: Lifespan stage that timed out ("startup" or "shutdown")
        """

class LifespanFailureError(Exception):
    """
    Raised when ASGI lifespan events fail.
    
    This exception indicates that an ASGI application's
    lifespan startup or shutdown event completed with
    an error or unexpected state.
    """
    
    def __init__(self, stage: str, message: str) -> None:
        """
        Initialize lifespan failure error.
        
        Args:
            stage: Lifespan stage that failed ("startup" or "shutdown")
            message: Error message describing the failure
        """

class UnexpectedMessageError(Exception):
    """
    Raised for unexpected ASGI messages.
    
    This exception occurs when an ASGI application sends
    messages that don't conform to the expected protocol
    sequence or message format.
    """
    
    def __init__(self, state: Enum, message_type: str) -> None:
        """
        Initialize unexpected message error.
        
        Args:
            state: Current protocol state when error occurred
            message_type: Type of unexpected message received
        """

class FrameTooLargeError(Exception):
    """
    Raised when protocol frame size exceeds limits.
    
    This exception indicates that a protocol frame (HTTP/2,
    WebSocket, etc.) exceeds the configured maximum size
    limits, potentially indicating a malicious request.
    """

from hypercorn.utils import load_application, wrap_app

# Load application from module path
try:
    app = load_application("myproject.wsgi:application", wsgi_max_body_size=16*1024*1024)
    print(f"Loaded application: {app}")
except Exception as e:
    print(f"Failed to load application: {e}")

# Wrap application automatically
wrapped_app = wrap_app(app, wsgi_max_body_size=16*1024*1024)
print(f"Wrapped application: {wrapped_app}")

# Check application type
from hypercorn.utils import is_asgi
if is_asgi(app):
    print("Application is ASGI-compatible")
else:
    print("Application is WSGI-compatible")

import time
from hypercorn.utils import files_to_watch, check_for_updates

# Get files to monitor
watch_files = files_to_watch()
print(f"Monitoring {len(watch_files)} files for changes")

# Check for changes periodically
while True:
    if check_for_updates(watch_files):
        print("Files changed - restarting server")
        # Trigger server restart
        break
    time.sleep(1)

import os
from hypercorn.utils import write_pid_file

# Write PID file for process management
try:
    write_pid_file("/var/run/hypercorn.pid")
    print(f"PID {os.getpid()} written to file")
except Exception as e:
    print(f"Failed to write PID file: {e}")

import socket
from hypercorn.utils import parse_socket_addr, repr_socket_addr

# Parse different address types
ipv4_addr = ("192.168.1.100", 8000)
ipv6_addr = ("::1", 8000, 0, 0)

parsed_ipv4 = parse_socket_addr(socket.AF_INET, ipv4_addr)
parsed_ipv6 = parse_socket_addr(socket.AF_INET6, ipv6_addr)

# Create string representations
ipv4_str = repr_socket_addr(socket.AF_INET, ipv4_addr)
ipv6_str = repr_socket_addr(socket.AF_INET6, ipv6_addr)

print(f"IPv4: {ipv4_str}")  # "192.168.1.100:8000"
print(f"IPv6: {ipv6_str}")  # "[::1]:8000"

from hypercorn.utils import build_and_validate_headers, filter_pseudo_headers

# Build headers from various formats
headers_dict = {"content-type": "text/html", "server": "hypercorn"}
headers_list = [("content-type", "text/html"), ("server", "hypercorn")]

validated_headers = build_and_validate_headers(headers_dict)
print(f"Validated headers: {validated_headers}")

# Filter HTTP/2 pseudo-headers
http2_headers = [
    (b":method", b"GET"),
    (b":path", b"/test"), 
    (b"user-agent", b"test-client"),
    (b":authority", b"example.com")
]

filtered = filter_pseudo_headers(http2_headers)
print(f"Filtered headers: {filtered}")  # Only user-agent remains

from hypercorn.utils import suppress_body

# Check if body should be suppressed
cases = [
    ("HEAD", 200),    # HEAD requests - suppress
    ("GET", 204),     # No Content - suppress  
    ("GET", 304),     # Not Modified - suppress
    ("GET", 200),     # Normal GET - don't suppress
    ("POST", 201),    # Normal POST - don't suppress
]

for method, status in cases:
    should_suppress = suppress_body(method, status)
    print(f"{method} {status}: {'suppress' if should_suppress else 'include'} body")

from hypercorn.config import Config
from hypercorn.utils import valid_server_name

# Configure server names
config = Config()
config.server_names = ["example.com", "*.example.com", "api.service.local"]

# Mock request object (simplified)
class MockRequest:
    def __init__(self, host):
        self.headers = {"host": host}

# Validate different server names
test_hosts = ["example.com", "api.example.com", "invalid.com", "api.service.local"]

for host in test_hosts:
    request = MockRequest(host)
    is_valid = valid_server_name(config, request)
    print(f"{host}: {'valid' if is_valid else 'invalid'}")