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

Core Application

Name: tessl/pypi-sanic
Author: tessl

The Sanic application class serves as the main entry point for creating web applications. It provides routing, middleware management, server lifecycle control, and all core functionality needed to build high-performance web services.

Capabilities

Application Initialization

Create a new Sanic application instance with customizable configuration and behavior options.

class Sanic:
    def __init__(
        self,
        name: str,
        config: Optional[Config] = None,
        ctx: Optional[Any] = None,
        router: Optional[Router] = None,
        signal_router: Optional[SignalRouter] = None,
        error_handler: Optional[ErrorHandler] = None,
        env_prefix: Optional[str] = "SANIC_",
        request_class: Optional[type[Request]] = None,
        strict_slashes: bool = False,
        log_config: Optional[dict[str, Any]] = None,
        configure_logging: bool = True,
        dumps: Optional[Callable[..., AnyStr]] = None,
        loads: Optional[Callable[..., Any]] = None,
        inspector: bool = False,
        inspector_class: Optional[type[Inspector]] = None,
        certloader_class: Optional[type[CertLoader]] = None,
    ):
        """
        Initialize a Sanic application.

        Parameters:
        - name: Application name (used for logging and identification)
        - config: Custom configuration instance
        - ctx: Application context object
        - router: Custom router instance
        - signal_router: Custom signal router instance
        - error_handler: Custom error handler instance
        - env_prefix: Prefix for environment variables (default: "SANIC_")
        - request_class: Custom request class
        - strict_slashes: Enforce strict slash handling
        - log_config: Logging configuration dictionary
        - configure_logging: Whether to configure logging automatically
        - dumps: JSON serialization function
        - loads: JSON deserialization function
        - inspector: Enable application inspector
        - inspector_class: Custom inspector class
        - certloader_class: Custom certificate loader class
        """

HTTP Route Decorators

Define HTTP endpoints using method-specific decorators for clean, readable route definitions.

def route(
    self,
    uri: str,
    methods: list = None,
    host: str = None,
    strict_slashes: bool = None,
    stream: bool = False,
    version: int = None,
    name: str = None,
    ignore_body: bool = False,
    apply: list = None,
    subprotocols: list = None,
    websocket: bool = False,
    unquote: bool = False,
    static: bool = False,
    version_prefix: str = "/v",
    error_format: str = None,
    ctx: Any = None,
):
    """
    Decorator for defining HTTP routes.

    Parameters:
    - uri: Route path with optional parameters
    - methods: HTTP methods (defaults to ['GET'])
    - host: Host restriction pattern
    - strict_slashes: Override global strict slash setting
    - stream: Enable request streaming
    - version: API version number
    - name: Route name for url_for
    - ignore_body: Don't parse request body
    - apply: List of decorators to apply
    - subprotocols: WebSocket subprotocols
    - websocket: Mark as WebSocket route
    - unquote: URL decode path parameters
    - static: Mark as static file route
    - version_prefix: Version URL prefix
    - error_format: Error response format
    - ctx: Route-specific context
    """

def get(self, uri: str, **kwargs):
    """GET method decorator."""

def post(self, uri: str, **kwargs):
    """POST method decorator."""

def put(self, uri: str, **kwargs):
    """PUT method decorator."""

def delete(self, uri: str, **kwargs):
    """DELETE method decorator."""

def patch(self, uri: str, **kwargs):
    """PATCH method decorator."""

def head(self, uri: str, **kwargs):
    """HEAD method decorator."""

def options(self, uri: str, **kwargs):
    """OPTIONS method decorator."""

Programmatic Routing

Add routes programmatically without decorators for dynamic route creation and advanced routing scenarios.

def add_route(
    self,
    handler,
    uri: str,
    methods: list = None,
    **kwargs
):
    """
    Add a route programmatically.

    Parameters:
    - handler: Handler function
    - uri: Route path
    - methods: HTTP methods list
    - **kwargs: Additional route options
    """

def add_websocket_route(
    self,
    handler,
    uri: str,
    **kwargs
):
    """
    Add a WebSocket route programmatically.

    Parameters:
    - handler: WebSocket handler function
    - uri: WebSocket path
    - **kwargs: Additional route options
    """

Static File Serving

Serve static files and directories with customizable options for caching, security, and performance.

def static(
    self,
    uri: str,
    file_or_directory: str,
    pattern: str = r"/?.+",
    use_modified_since: bool = True,
    use_content_range: bool = False,
    stream_large_files: bool = False,
    name: str = "static",
    host: str = None,
    strict_slashes: bool = None,
    content_type: str = None,
    apply: list = None,
    resource_type: str = "file",
    index: str = None,
    directory_view: bool = False,
    directory_handler: Callable = None,
):
    """
    Serve static files or directories.

    Parameters:
    - uri: URL path for static files
    - file_or_directory: Path to file or directory
    - pattern: File matching pattern
    - use_modified_since: Enable conditional requests
    - use_content_range: Enable range requests
    - stream_large_files: Stream large files
    - name: Route name
    - host: Host restriction
    - strict_slashes: Slash handling
    - content_type: Override content type
    - apply: Decorators to apply
    - resource_type: Resource type identifier
    - index: Directory index file
    - directory_view: Enable directory listing
    - directory_handler: Custom directory handler
    """

WebSocket Routes

Define WebSocket endpoints for real-time bidirectional communication.

def websocket(
    self,
    uri: str,
    host: str = None,
    strict_slashes: bool = None,
    subprotocols: list = None,
    name: str = None,
    apply: list = None,
    version: int = None,
    version_prefix: str = "/v",
    error_format: str = None,
):
    """
    Decorator for WebSocket routes.

    Parameters:
    - uri: WebSocket path
    - host: Host restriction
    - strict_slashes: Slash handling
    - subprotocols: Supported subprotocols
    - name: Route name
    - apply: Decorators to apply
    - version: API version
    - version_prefix: Version prefix
    - error_format: Error format
    """

Server Lifecycle

Control server startup, shutdown, and runtime behavior with comprehensive configuration options.

def run(
    self,
    host: str = "127.0.0.1",
    port: int = 8000,
    debug: bool = False,
    ssl: dict = None,
    sock: socket = None,
    workers: int = 1,
    protocol: Type = None,
    backlog: int = 100,
    stop_event: 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: Stop event for graceful shutdown
    - register_sys_signals: Register system signals
    - run_multiple: Run multiple workers
    - run_async: Run asynchronously
    - connections: Connection set
    - signal: Signal object
    - state: Shared state dictionary
    - asyncio_server_kwargs: Additional asyncio server arguments
    - return_asyncio_server: Return server object
    """

def create_server(
    self,
    host: str = "127.0.0.1",
    port: int = 8000,
    **kwargs
):
    """
    Create server instance without running.

    Returns:
    Server instance
    """

def stop(self):
    """Stop the running server."""

def restart(self, **kwargs):
    """Restart the server with new configuration."""

Event Listeners

def listener(self, event: str):
    """
    Decorator for event listeners.

    Parameters:
    - event: Event name ('before_server_start', 'after_server_start', 
             'before_server_stop', 'after_server_stop')
    """

def before_server_start(self, listener: Callable):
    """Register before_server_start listener."""

def after_server_start(self, listener: Callable):
    """Register after_server_start listener."""

def before_server_stop(self, listener: Callable):
    """Register before_server_stop listener."""

def after_server_stop(self, listener: Callable):
    """Register after_server_stop listener."""

def main_process_start(self, listener: Callable):
    """Register main_process_start listener."""

def main_process_ready(self, listener: Callable):
    """Register main_process_ready listener."""

def main_process_stop(self, listener: Callable):
    """Register main_process_stop listener."""

def reload_process_start(self, listener: Callable):
    """Register reload_process_start listener."""

def reload_process_stop(self, listener: Callable):
    """Register reload_process_stop listener."""

def before_reload_trigger(self, listener: Callable):
    """Register before_reload_trigger listener."""

def after_reload_trigger(self, listener: Callable):
    """Register after_reload_trigger listener."""

URL Generation and Testing

Generate URLs for named routes and create test clients for application testing.

def url_for(
    self,
    view_name: str,
    **kwargs
):
    """
    Generate URL for named route.

    Parameters:
    - view_name: Route name
    - **kwargs: Route parameters

    Returns:
    Generated URL string
    """

def test_client(self, **kwargs):
    """
    Create test client for application testing.

    Returns:
    Test client instance
    """

def asgi_client(self, **kwargs):
    """
    Create ASGI test client.

    Returns:
    ASGI client instance
    """

Middleware Management

def middleware(
    self,
    middleware_or_request: Union[MiddlewareType, str],
    attach_to: str = "request",
    apply: bool = True,
    *,
    priority: int = 0,
):
    """
    Decorator or method for registering middleware.
    
    Parameters:
    - middleware_or_request: Middleware function or attachment point
    - attach_to: Attach to 'request' or 'response'
    - apply: Whether to apply the middleware
    - priority: Middleware priority order
    """

def register_middleware(
    self,
    middleware: Union[MiddlewareType, Middleware],
    attach_to: str = "request",
    *,
    priority: Union[Default, int] = _default,
):
    """
    Register middleware programmatically.
    
    Parameters:
    - middleware: Middleware function or instance
    - attach_to: Attach to 'request' or 'response'
    - priority: Middleware priority order
    """

Exception Handling

def exception(
    self,
    *exceptions: Union[type[Exception], Exception],
    apply: bool = True,
):
    """
    Decorator for exception handlers.
    
    Parameters:
    - exceptions: Exception types to handle
    - apply: Whether to apply the handler
    """

Signal System

def signal(
    self,
    event: Union[str, Enum],
    *,
    apply: bool = True,
    condition: Optional[dict[str, Any]] = None,
    exclusive: bool = True,
    priority: int = 0,
):
    """
    Decorator for signal handlers.
    
    Parameters:
    - event: Signal event name
    - apply: Whether to apply the handler
    - condition: Signal condition dictionary
    - exclusive: Whether signal is exclusive
    - priority: Handler priority
    """

async def dispatch(
    self,
    event: str,
    *,
    condition: Optional[dict[str, str]] = None,
    context: Optional[dict[str, Any]] = None,
    fail_not_found: bool = True,
    inline: bool = False,
    reverse: bool = False,
):
    """
    Dispatch a signal event.
    
    Parameters:
    - event: Event name to dispatch
    - condition: Event condition
    - context: Event context data
    - fail_not_found: Fail if no handlers found
    - inline: Execute inline
    - reverse: Execute in reverse order
    """

Blueprint Registration

def blueprint(
    self,
    blueprint: Union[Blueprint, Iterable[Blueprint], BlueprintGroup],
    *,
    url_prefix: Optional[str] = None,
    version: Optional[Union[int, float, str]] = None,
    strict_slashes: Optional[bool] = None,
    version_prefix: Optional[str] = None,
    name_prefix: Optional[str] = None,
):
    """
    Register one or more blueprints.
    
    Parameters:
    - blueprint: Blueprint(s) to register
    - url_prefix: URL prefix for blueprint routes
    - version: API version for blueprint
    - strict_slashes: Strict slash handling
    - version_prefix: Version prefix format
    - name_prefix: Name prefix for routes
    """

Application Properties

Access application configuration, context, routing, and other runtime information.

@property
def config(self) -> Config:
    """Application configuration object."""

@property
def ctx(self):
    """Application context object."""

@property
def router(self):
    """Application router instance."""

@property
def blueprints(self) -> dict:
    """Registered blueprints dictionary."""

@property
def listeners(self) -> dict:
    """Event listeners dictionary."""

@property
def middleware(self) -> list:
    """Middleware stack list."""

@property
def request_middleware(self) -> list:
    """Request middleware list."""

@property
def response_middleware(self) -> list:
    """Response middleware list."""

@property
def exception_middleware(self) -> list:
    """Exception middleware list."""

@property
def error_handler(self):
    """Application error handler."""

@property
def name(self) -> str:
    """Application name."""

@property
def strict_slashes(self) -> bool:
    """Strict slash handling setting."""

@property
def is_running(self) -> bool:
    """Whether server is currently running."""

@property
def is_request_stream(self) -> bool:
    """Whether request streaming is enabled."""

Usage Examples

Basic Application Setup

from sanic import Sanic
from sanic.response import json

app = Sanic("MyApplication")

@app.route("/api/users/<user_id:int>")
async def get_user(request, user_id):
    # Fetch user from database
    user = await fetch_user(user_id)
    return json({"user": user})

@app.route("/api/users", methods=["POST"])
async def create_user(request):
    user_data = request.json
    user = await create_user(user_data)
    return json({"user": user}, status=201)

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

Advanced Configuration

from sanic import Sanic
from sanic.config import Config

# Custom configuration
config = Config()
config.REQUEST_TIMEOUT = 60
config.RESPONSE_TIMEOUT = 60
config.KEEP_ALIVE_TIMEOUT = 5

app = Sanic("MyApp", config=config)

# Environment-based configuration
app.config.load_environment_vars("MYAPP_")

# SSL configuration
ssl_config = {
    "cert": "/path/to/cert.pem",
    "key": "/path/to/key.pem"
}

app.run(host="0.0.0.0", port=8443, ssl=ssl_config)

Event Listeners and Lifecycle

@app.before_server_start
async def setup_database(app, loop):
    """Initialize database connection."""
    app.ctx.db = await create_database_connection()

@app.after_server_start
async def notify_ready(app, loop):
    """Notify that server is ready."""
    print(f"Server ready at http://{app.config.HOST}:{app.config.PORT}")

@app.before_server_stop
async def cleanup(app, loop):
    """Clean up resources."""
    await app.ctx.db.close()

Install with Tessl CLI