Server Management

from uvicorn import Server, Config
from uvicorn.server import ServerState, HANDLED_SIGNALS

class Server:
    """
    ASGI server implementation.

    Manages the asyncio event loop, socket binding, protocol instances,
    signal handling, and graceful shutdown for ASGI applications.
    """

    def __init__(self, config: Config) -> None:
        """
        Initialize server with configuration.

        Args:
            config: Server configuration object

        Attributes set:
            config: Server configuration
            server_state: Shared state across protocol connections
            started: Whether server has started (False initially)
            should_exit: Signal for graceful shutdown (False initially)
            force_exit: Signal for immediate shutdown (False initially)
            last_notified: Timestamp of last notification (0.0 initially)
        """

    def run(self, sockets: list[socket.socket] | None = None) -> None:
        """
        Run the server (blocking).

        Creates an event loop and runs the server synchronously until
        shutdown. This is the main entry point for running the server.

        **NOTE**: This method automatically calls config.load() if not already loaded,
        so you don't need to call it manually when using run().

        Args:
            sockets: Optional pre-bound sockets to use instead of binding new ones

        This method:
        1. Configures logging
        2. Loads the application configuration (calls config.load() if needed)
        3. Creates an event loop
        4. Runs serve() in the event loop
        5. Handles cleanup on exit
        """

    async def serve(self, sockets: list[socket.socket] | None = None) -> None:
        """
        Run the server (async).

        Asynchronous entry point for running the server. Can be used
        when you need to run the server within an existing event loop.

        **NOTE**: This method automatically calls config.load() if not already loaded,
        so you don't need to call it manually when using serve().

        Args:
            sockets: Optional pre-bound sockets to use instead of binding new ones

        This method:
        1. Loads config if not already loaded (calls config.load() if needed)
        2. Calls startup()
        3. Runs main_loop() if startup succeeded
        4. Calls shutdown()
        """

    async def startup(self, sockets: list[socket.socket] | None = None) -> None:
        """
        Start the server.

        **IMPORTANT**: Before calling startup(), you must call config.load() to initialize
        the configuration. The startup() method requires access to config.lifespan_class
        and other attributes that are only set during config.load().

        Performs server initialization:
        - Starts lifespan handler
        - Binds sockets
        - Creates asyncio server
        - Starts accepting connections

        Args:
            sockets: Optional pre-bound sockets to use

        Sets:
            started: True if startup succeeds
            should_exit: True if startup fails

        Raises:
            AttributeError: If config.load() has not been called before startup()
        """

    async def shutdown(self, sockets: list[socket.socket] | None = None) -> None:
        """
        Shutdown the server gracefully.

        Performs graceful shutdown sequence:
        1. Stops accepting new connections
        2. Waits for active connections to complete (up to timeout_graceful_shutdown)
        3. Shuts down lifespan handler
        4. Closes all remaining connections
        5. Cancels all background tasks

        Args:
            sockets: Optional sockets to close

        This method ensures all resources are cleaned up properly.
        """

    async def main_loop(self) -> None:
        """
        Main server event loop.

        Runs a periodic tick loop that:
        - Checks for shutdown signals
        - Performs periodic maintenance
        - Monitors connection health

        The loop continues until should_exit is set.
        """

    async def on_tick(self, counter: int) -> bool:
        """
        Periodic tick callback.

        Called regularly from main_loop() for periodic tasks.

        Args:
            counter: Tick counter (increments each tick)

        Returns:
            True to continue running, False to trigger shutdown
        """

    def handle_exit(self, sig: int, frame: types.FrameType | None) -> None:
        """
        Signal handler for shutdown signals.

        Handles SIGINT, SIGTERM, and SIGBREAK (Windows) to trigger
        graceful shutdown.

        Args:
            sig: Signal number
            frame: Current stack frame

        Sets should_exit on first signal, force_exit on second signal.
        """

    def capture_signals(self) -> Generator[None, None, None]:
        """
        Context manager for signal handling.

        Sets up signal handlers on entry and restores original handlers
        on exit. Used internally to manage signal handling during server
        execution.

        Yields:
            None

        Example usage (internal):
            with server.capture_signals():
                # Server running with signal handlers active
                ...
            # Original signal handlers restored
        """

class ServerState:
    """
    Shared state available across all protocol instances.

    Provides coordination and statistics tracking for all active
    connections and background tasks.
    """

    total_requests: int
    """Total number of HTTP requests processed since server start."""

    connections: set[asyncio.Protocol]
    """
    Set of active protocol connection instances.

    Each connection is a protocol instance (HTTP or WebSocket) that is
    currently handling a client connection. Connections add themselves
    to this set when created and remove themselves when closed.
    """

    tasks: set[asyncio.Task[None]]
    """
    Set of active background tasks.

    Background tasks created by protocol handlers for async operations.
    Tasks add themselves to this set when created and remove themselves
    when complete.
    """

    default_headers: list[tuple[bytes, bytes]]
    """
    Default HTTP headers to include in all responses.

    Headers are stored as byte tuples and added to every HTTP response
    unless overridden by the application. Includes Server and Date headers
    if configured, plus any custom headers from config.headers.
    """

# Signals that the server handles for graceful shutdown
HANDLED_SIGNALS: tuple[int, ...]
"""
Signals handled by the server for graceful shutdown.

On Unix: (signal.SIGINT, signal.SIGTERM)
On Windows: (signal.SIGINT, signal.SIGTERM, signal.SIGBREAK)

First signal triggers graceful shutdown (should_exit = True).
Second signal triggers immediate shutdown (force_exit = True).
"""

from uvicorn import Config, Server

# Define ASGI application
async def app(scope, receive, send):
    assert scope['type'] == 'http'
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/plain']],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, World!',
    })

# Create configuration
config = Config(app=app, host="127.0.0.1", port=8000)

# Create and run server
server = Server(config)
server.run()

import asyncio
from uvicorn import Config, Server

async def app(scope, receive, send):
    # Your ASGI application
    ...

async def main():
    config = Config(app=app, host="127.0.0.1", port=8000)
    server = Server(config)

    # Run server within existing event loop
    await server.serve()

# Run with asyncio
asyncio.run(main())

import socket
from uvicorn import Config, Server

# Create and bind socket manually
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(("127.0.0.1", 8000))

# Pass socket to server
config = Config(app="myapp:app")
server = Server(config)
server.run(sockets=[sock])

import asyncio
from uvicorn import Config, Server

async def app(scope, receive, send):
    # Your ASGI application
    ...

async def run_server():
    config = Config(
        app=app,
        host="127.0.0.1",
        port=8000,
        timeout_graceful_shutdown=30,  # 30 second graceful shutdown
    )

    # IMPORTANT: Must call load() before creating Server and calling startup()
    config.load()

    server = Server(config)

    # Start server
    await server.startup()

    if server.started:
        try:
            # Run until interrupted
            await server.main_loop()
        finally:
            # Ensure clean shutdown
            await server.shutdown()

asyncio.run(run_server())

import asyncio
from uvicorn import Config, Server

async def app(scope, receive, send):
    # Your ASGI application
    ...

async def run_with_timeout():
    config = Config(app=app, host="127.0.0.1", port=8000)
    config.load()  # Required before calling startup()

    server = Server(config)

    # Start server
    await server.startup()

    if server.started:
        # Run server for 60 seconds then shutdown
        try:
            await asyncio.sleep(60)
        finally:
            server.should_exit = True
            await server.shutdown()

asyncio.run(run_with_timeout())

import asyncio
from uvicorn import Config, Server

async def app(scope, receive, send):
    # Your ASGI application
    ...

async def monitor_server():
    config = Config(app=app, host="127.0.0.1", port=8000)
    config.load()  # Required before calling startup()

    server = Server(config)

    await server.startup()

    if server.started:
        # Monitor connections and requests
        async def stats_reporter():
            while not server.should_exit:
                state = server.server_state
                print(f"Active connections: {len(state.connections)}")
                print(f"Total requests: {state.total_requests}")
                print(f"Background tasks: {len(state.tasks)}")
                await asyncio.sleep(5)

        # Run stats reporter alongside server
        stats_task = asyncio.create_task(stats_reporter())

        try:
            await server.main_loop()
        finally:
            stats_task.cancel()
            await server.shutdown()

asyncio.run(monitor_server())

import signal
import asyncio
from uvicorn import Config, Server

async def app(scope, receive, send):
    # Your ASGI application
    ...

async def run_with_custom_signals():
    config = Config(app=app, host="127.0.0.1", port=8000)
    config.load()  # Required before calling startup()

    server = Server(config)

    # Custom shutdown handler
    def shutdown_handler(signum, frame):
        print(f"Received signal {signum}, shutting down...")
        server.should_exit = True

    # Set up custom signal handlers
    signal.signal(signal.SIGINT, shutdown_handler)
    signal.signal(signal.SIGTERM, shutdown_handler)

    await server.startup()

    if server.started:
        try:
            await server.main_loop()
        finally:
            await server.shutdown()

asyncio.run(run_with_custom_signals())

import asyncio
from uvicorn import Config, Server

async def app(scope, receive, send):
    # Your ASGI application
    ...

async def run_with_stages():
    config = Config(app=app, host="127.0.0.1", port=8000)

    # Stage 0: Load configuration (required before startup)
    print("Loading configuration...")
    config.load()

    server = Server(config)

    # Stage 1: Startup
    print("Starting server...")
    await server.startup()

    if not server.started:
        print("Server failed to start")
        return

    print(f"Server started on {config.host}:{config.port}")
    print(f"Active connections: {len(server.server_state.connections)}")

    # Stage 2: Run
    try:
        await server.main_loop()
    except KeyboardInterrupt:
        print("\nInterrupted by user")

    # Stage 3: Shutdown
    print("Shutting down server...")
    await server.shutdown()
    print("Server stopped")

asyncio.run(run_with_stages())

from uvicorn import Config, Server

async def app(scope, receive, send):
    # Your ASGI application
    ...

# Limit concurrent connections
config = Config(
    app=app,
    host="0.0.0.0",
    port=8000,
    limit_concurrency=100,  # Max 100 concurrent connections
    backlog=2048,  # Socket backlog
)

server = Server(config)
server.run()

from uvicorn import Config, Server

async def app(scope, receive, send):
    # Your ASGI application
    ...

# Restart after processing 10000 requests
config = Config(
    app=app,
    host="0.0.0.0",
    port=8000,
    limit_max_requests=10000,
)

server = Server(config)
server.run()