tessl/pypi-sanic

A modern Python web server and web framework designed for high performance and speed using async/await syntax.

—

Pending

Overview

Eval results

Files

Server and Deployment

Name: tessl/pypi-sanic
Author: tessl

Sanic provides comprehensive server configuration and deployment options for production environments. This includes multi-worker support, SSL/TLS configuration, various server protocols, process management, and production-ready features.

Capabilities

Server Execution

Core methods for running Sanic applications with various configuration options.

def run(
    host: str = "127.0.0.1",
    port: int = 8000,
    debug: bool = False,
    ssl: dict = None,
    sock: socket.socket = None,
    workers: int = 1,
    protocol: type = None,
    backlog: int = 100,
    stop_event: asyncio.Event = None,
    register_sys_signals: bool = True,
    run_multiple: bool = False,
    run_async: bool = False,
    connections: set = None,
    signal: Signal = None,
    state: dict = None,
    asyncio_server_kwargs: dict = None,
    return_asyncio_server: bool = False,
    **kwargs
):
    """
    Run the Sanic server.
    
    Parameters:
    - host: Server host address
    - port: Server port number
    - debug: Enable debug mode
    - ssl: SSL configuration dictionary
    - sock: Custom socket object
    - workers: Number of worker processes
    - protocol: Custom protocol class
    - backlog: Connection backlog size
    - stop_event: Event for graceful shutdown
    - register_sys_signals: Register system signal handlers
    - run_multiple: Enable multi-worker mode
    - run_async: Run server asynchronously
    - connections: Connection tracking set
    - signal: Signal object for coordination
    - state: Shared state dictionary
    - asyncio_server_kwargs: Additional asyncio server arguments
    - return_asyncio_server: Return server object instead of running
    
    Returns:
    - None (runs server) or server object (if return_asyncio_server=True)
    """

def create_server(
    host: str = "127.0.0.1",
    port: int = 8000,
    return_asyncio_server: bool = False,
    asyncio_server_kwargs: dict = None,
    **kwargs
):
    """
    Create server instance without running it.
    
    Parameters:
    - host: Server host address
    - port: Server port number
    - return_asyncio_server: Return asyncio server object
    - asyncio_server_kwargs: Additional asyncio server arguments
    
    Returns:
    - Server instance or asyncio server object
    """

async def serve(
    host: str = "127.0.0.1",
    port: int = 8000,
    ssl: dict = None,
    sock: socket.socket = None,
    protocol: type = None,
    backlog: int = 100,
    connections: set = None,
    **kwargs
):
    """
    Serve the application asynchronously.
    
    Parameters:
    - host: Server host address
    - port: Server port number
    - ssl: SSL configuration
    - sock: Custom socket
    - protocol: Custom protocol class
    - backlog: Connection backlog
    - connections: Connection tracking
    
    Returns:
    - Server object
    """

Multi-Worker Deployment

Scale applications horizontally using multiple worker processes.

def run_multiple(
    workers: int,
    host: str = "127.0.0.1",
    port: int = 8000,
    ssl: dict = None,
    sock: socket.socket = None,
    **kwargs
):
    """
    Run server with multiple worker processes.
    
    Parameters:
    - workers: Number of worker processes
    - host: Server host address
    - port: Server port number
    - ssl: SSL configuration
    - sock: Custom socket
    
    Features:
    - Process-based parallelism
    - Shared socket for load balancing
    - Graceful worker management
    - Signal handling for coordinated shutdown
    """

# Worker configuration options
WORKERS: int = 1                    # Number of worker processes
WORKER_CLASS: str = "sanic.worker.GunicornWorker"  # Worker class
WORKER_CONNECTIONS: int = 1000      # Max connections per worker
WORKER_MAX_REQUESTS: int = 0        # Max requests before worker restart
WORKER_MAX_REQUESTS_JITTER: int = 0 # Random jitter for max requests

SSL/TLS Configuration

Secure server deployment with SSL/TLS encryption.

# SSL configuration dictionary structure
ssl_config = {
    "cert": str,                    # Certificate file path
    "key": str,                     # Private key file path
    "ca_certs": str,               # CA certificates file path (optional)
    "ciphers": str,                # Cipher suites (optional)
    "ssl_version": int,            # SSL/TLS version (optional)
    "do_handshake_on_connect": bool, # Handshake behavior (optional)
    "check_hostname": bool,        # Hostname verification (optional)
    "certfile": str,               # Alias for cert (optional)
    "keyfile": str,                # Alias for key (optional)
    "password": str,               # Private key password (optional)
}

# Example SSL configurations
ssl_basic = {
    "cert": "/path/to/certificate.pem",
    "key": "/path/to/private_key.pem"
}

ssl_advanced = {
    "cert": "/path/to/certificate.pem",
    "key": "/path/to/private_key.pem",
    "ca_certs": "/path/to/ca_certificates.pem",
    "ciphers": "ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20:!aNULL:!MD5:!DSS",
    "ssl_version": 2,  # ssl.PROTOCOL_TLS
    "check_hostname": False
}

Unix Socket Support

Deploy applications using Unix domain sockets for inter-process communication.

def run_unix_socket(
    path: str,
    mode: int = 0o666,
    **kwargs
):
    """
    Run server on Unix domain socket.
    
    Parameters:
    - path: Unix socket file path
    - mode: Socket file permissions
    
    Benefits:
    - Lower overhead than TCP sockets
    - Better security (filesystem permissions)
    - Ideal for reverse proxy setups
    """

# Configuration for Unix sockets
UNIX: str = "/tmp/sanic.sock"       # Unix socket path
UNIX_MODE: int = 0o666              # Socket file permissions

Server Protocols

Support for different server protocols and implementations.

# Built-in protocol classes
from sanic.server.protocols.http_protocol import HttpProtocol
from sanic.server.protocols.websocket_protocol import WebSocketProtocol

class CustomHttpProtocol(HttpProtocol):
    """Custom HTTP protocol implementation."""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # Custom protocol initialization
    
    async def http_request_handler(self, request):
        """Custom request handling logic."""
        # Custom processing
        return await super().http_request_handler(request)

# Use custom protocol
app.run(protocol=CustomHttpProtocol)

Process Management

Server lifecycle management and graceful shutdown handling.

def stop(self):
    """
    Stop the running server gracefully.
    
    Features:
    - Graceful connection closure
    - Worker process coordination
    - Resource cleanup
    - Signal handling
    """

def restart(self, **kwargs):
    """
    Restart the server with new configuration.
    
    Parameters:
    - **kwargs: New configuration options
    
    Features:
    - Zero-downtime restart capability
    - Configuration updates
    - Worker process cycling
    """

# Graceful shutdown configuration
GRACEFUL_SHUTDOWN_TIMEOUT: float = 15.0  # Shutdown timeout in seconds
KEEP_ALIVE_TIMEOUT: int = 5               # Keep-alive connection timeout
REQUEST_TIMEOUT: int = 60                 # Request processing timeout
RESPONSE_TIMEOUT: int = 60                # Response sending timeout

Development Server Features

Development-focused server features for rapid development and debugging.

# Development configuration
DEBUG: bool = False                 # Debug mode
AUTO_RELOAD: bool = False          # Auto-reload on file changes
ACCESS_LOG: bool = True            # Enable access logging
MOTD: bool = True                  # Show message of the day
LOGO: str = None                   # Custom logo display

# Auto-reload functionality
def enable_auto_reload(app, reload_dir: str = "."):
    """
    Enable automatic server restart on file changes.
    
    Parameters:
    - app: Sanic application instance
    - reload_dir: Directory to watch for changes
    
    Features:
    - File system monitoring
    - Automatic server restart
    - Development workflow optimization
    """

Production Deployment Patterns

Recommended patterns and configurations for production environments.

# Production server configuration
production_config = {
    "host": "0.0.0.0",
    "port": 8000,
    "workers": 4,                   # CPU core count
    "debug": False,
    "access_log": True,
    "auto_reload": False,
    "backlog": 2048,
    "keep_alive_timeout": 5,
    "request_timeout": 60,
    "response_timeout": 60,
    "request_max_size": 100 * 1024 * 1024,  # 100MB
}

# Load balancer configuration
load_balancer_config = {
    "host": "127.0.0.1",           # Bind to localhost
    "port": 8000,
    "workers": 8,
    "backlog": 4096,
    "forwarded_secret": "your-secret-key",
    "real_ip_header": "X-Real-IP",
    "proxies_count": 1,
}

Usage Examples

Basic Server Deployment

from sanic import Sanic
from sanic.response import json

app = Sanic("ProductionApp")

@app.route("/health")
async def health_check(request):
    return json({"status": "healthy", "service": "ProductionApp"})

@app.route("/api/data")
async def get_data(request):
    return json({"data": "production data"})

if __name__ == "__main__":
    # Production server configuration
    app.run(
        host="0.0.0.0",
        port=8000,
        workers=4,
        debug=False,
        access_log=True
    )

SSL/HTTPS Deployment

import ssl
from sanic import Sanic

app = Sanic("SecureApp")

@app.route("/")
async def secure_endpoint(request):
    return json({
        "message": "Secure connection established",
        "scheme": request.scheme,
        "secure": request.scheme == "https"
    })

if __name__ == "__main__":
    # SSL configuration
    ssl_config = {
        "cert": "/etc/ssl/certs/server.crt",
        "key": "/etc/ssl/private/server.key",
        "ssl_version": ssl.PROTOCOL_TLS,
        "ciphers": "ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20:!aNULL:!MD5:!DSS"
    }
    
    app.run(
        host="0.0.0.0",
        port=8443,
        ssl=ssl_config,
        workers=4
    )

Multi-Worker Production Setup

import os
import multiprocessing
from sanic import Sanic

app = Sanic("MultiWorkerApp")

# Production configuration
app.config.update({
    "REQUEST_TIMEOUT": 60,
    "RESPONSE_TIMEOUT": 60,
    "KEEP_ALIVE_TIMEOUT": 5,
    "REQUEST_MAX_SIZE": 100 * 1024 * 1024,  # 100MB
    "GRACEFUL_SHUTDOWN_TIMEOUT": 15.0,
})

@app.before_server_start
async def setup_production(app, loop):
    """Production setup tasks."""
    # Initialize database pool
    app.ctx.db_pool = await create_database_pool()
    
    # Setup logging
    setup_production_logging()
    
    # Initialize cache
    app.ctx.cache = await create_cache_client()

@app.before_server_stop
async def cleanup_production(app, loop):
    """Production cleanup tasks."""
    if hasattr(app.ctx, 'db_pool'):
        await app.ctx.db_pool.close()
    
    if hasattr(app.ctx, 'cache'):
        await app.ctx.cache.close()

if __name__ == "__main__":
    # Calculate optimal worker count
    worker_count = min(multiprocessing.cpu_count(), int(os.getenv("MAX_WORKERS", 8)))
    
    app.run(
        host="0.0.0.0",
        port=int(os.getenv("PORT", 8000)),
        workers=worker_count,
        debug=False,
        access_log=True,
        backlog=2048
    )

Unix Socket Deployment

import os
from sanic import Sanic

app = Sanic("UnixSocketApp")

@app.route("/status")
async def status(request):
    return json({"status": "running", "socket": "unix"})

if __name__ == "__main__":
    socket_path = "/tmp/sanic_app.sock"
    
    # Remove existing socket file
    if os.path.exists(socket_path):
        os.unlink(socket_path)
    
    # Run on Unix socket
    app.run(
        unix=socket_path,
        workers=4,
        debug=False
    )
    
    # Set socket permissions
    os.chmod(socket_path, 0o666)

Container Deployment

import signal
import sys
from sanic import Sanic

app = Sanic("ContainerApp")

# Container-friendly configuration
app.config.update({
    "HOST": "0.0.0.0",
    "PORT": int(os.getenv("PORT", 8000)),
    "WORKERS": int(os.getenv("WORKERS", 1)),
    "DEBUG": os.getenv("DEBUG", "false").lower() == "true",
    "ACCESS_LOG": os.getenv("ACCESS_LOG", "true").lower() == "true",
})

@app.before_server_start
async def setup_container(app, loop):
    """Container-specific setup."""
    app.logger.info("Starting application in container")
    
    # Health check endpoint for container orchestration
    @app.route("/health")
    async def health_check(request):
        return json({"status": "healthy"})
    
    @app.route("/ready")
    async def readiness_check(request):
        # Check dependencies (database, cache, etc.)
        ready = await check_dependencies()
        if ready:
            return json({"status": "ready"})
        else:
            return json({"status": "not ready"}, status=503)

def signal_handler(signum, frame):
    """Handle container shutdown signals."""
    app.logger.info(f"Received signal {signum}, shutting down gracefully")
    app.stop()

if __name__ == "__main__":
    # Register signal handlers for graceful shutdown
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)
    
    try:
        app.run(
            host=app.config.HOST,
            port=app.config.PORT,
            workers=app.config.WORKERS,
            debug=app.config.DEBUG,
            access_log=app.config.ACCESS_LOG
        )
    except KeyboardInterrupt:
        app.logger.info("Application stopped by user")
        sys.exit(0)

Reverse Proxy Configuration

from sanic import Sanic
from sanic.response import json

app = Sanic("ProxiedApp")

# Configure for reverse proxy setup
app.config.update({
    "FORWARDED_SECRET": os.getenv("FORWARDED_SECRET"),
    "REAL_IP_HEADER": "X-Real-IP",
    "PROXIES_COUNT": 1,  # Number of proxy layers
})

@app.middleware("request")
async def log_real_ip(request):
    """Log real client IP from proxy headers."""
    real_ip = request.ip  # Sanic handles proxy headers automatically
    app.logger.info(f"Request from {real_ip} to {request.path}")

@app.route("/proxy-info")
async def proxy_info(request):
    """Display proxy-related information."""
    return json({
        "client_ip": request.ip,
        "forwarded": dict(request.forwarded),
        "remote_addr": request.remote_addr,
        "scheme": request.scheme,
        "host": request.host,
        "headers": {
            "x-forwarded-for": request.headers.get("X-Forwarded-For"),
            "x-forwarded-proto": request.headers.get("X-Forwarded-Proto"),
            "x-real-ip": request.headers.get("X-Real-IP"),
        }
    })

if __name__ == "__main__":
    # Run behind reverse proxy (nginx, HAProxy, etc.)
    app.run(
        host="127.0.0.1",  # Bind to localhost only
        port=8000,
        workers=4,
        debug=False
    )

Custom Server Protocol

from sanic.server.protocols.http_protocol import HttpProtocol
from sanic import Sanic

class CustomProtocol(HttpProtocol):
    """Custom HTTP protocol with request logging."""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.request_count = 0
    
    async def http_request_handler(self, request):
        """Custom request handling with metrics."""
        self.request_count += 1
        
        # Add custom headers
        request.ctx.request_number = self.request_count
        
        # Log request
        self.logger.info(f"Processing request #{self.request_count}: {request.method} {request.path}")
        
        # Call parent handler
        response = await super().http_request_handler(request)
        
        # Add custom response headers
        if response:
            response.headers["X-Request-Number"] = str(self.request_count)
            response.headers["X-Server-Protocol"] = "custom"
        
        return response

app = Sanic("CustomProtocolApp")

@app.route("/")
async def index(request):
    return json({
        "message": "Custom protocol server",
        "request_number": request.ctx.request_number
    })

if __name__ == "__main__":
    app.run(
        host="0.0.0.0",
        port=8000,
        protocol=CustomProtocol,
        workers=2
    )

Install with Tessl CLI