Uvicorn

import uvicorn

from uvicorn import main, run, Config, Server

from uvicorn._types import (
    ASGIApplication,
    ASGI3Application,
    Scope,
    HTTPScope,
    WebSocketScope,
    ASGIReceiveCallable,
    ASGISendCallable,
)

import uvicorn

# Simple 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!',
    })

# Run the server programmatically
if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

import uvicorn

# Run an application defined in another module
if __name__ == "__main__":
    uvicorn.run("myapp:app", host="0.0.0.0", port=8000, reload=True)

def run(
    app: ASGIApplication | Callable | str,
    *,
    host: str = "127.0.0.1",
    port: int = 8000,
    uds: str | None = None,
    fd: int | None = None,
    loop: Literal["none", "auto", "asyncio", "uvloop"] | str = "auto",
    http: Literal["auto", "h11", "httptools"] | str | type[asyncio.Protocol] = "auto",
    ws: Literal["auto", "none", "websockets", "websockets-sansio", "wsproto"] | str | type[asyncio.Protocol] = "auto",
    ws_max_size: int = 16777216,
    ws_max_queue: int = 32,
    ws_ping_interval: float | None = 20.0,
    ws_ping_timeout: float | None = 20.0,
    ws_per_message_deflate: bool = True,
    lifespan: Literal["auto", "on", "off"] = "auto",
    interface: Literal["auto", "asgi3", "asgi2", "wsgi"] = "auto",
    reload: bool = False,
    reload_dirs: list[str] | str | None = None,
    reload_includes: list[str] | str | None = None,
    reload_excludes: list[str] | str | None = None,
    reload_delay: float = 0.25,
    workers: int | None = None,
    env_file: str | os.PathLike | None = None,
    log_config: dict[str, Any] | str | configparser.RawConfigParser | typing.IO[Any] | None = LOGGING_CONFIG,
    log_level: str | int | None = None,
    access_log: bool = True,
    proxy_headers: bool = True,
    server_header: bool = True,
    date_header: bool = True,
    forwarded_allow_ips: list[str] | str | None = None,
    root_path: str = "",
    limit_concurrency: int | None = None,
    backlog: int = 2048,
    limit_max_requests: int | None = None,
    timeout_keep_alive: int = 5,
    timeout_graceful_shutdown: int | None = None,
    timeout_worker_healthcheck: int = 5,
    ssl_keyfile: str | os.PathLike | None = None,
    ssl_certfile: str | os.PathLike | None = None,
    ssl_keyfile_password: str | None = None,
    ssl_version: int = ssl.PROTOCOL_TLS_SERVER,
    ssl_cert_reqs: int = ssl.CERT_NONE,
    ssl_ca_certs: str | os.PathLike | None = None,
    ssl_ciphers: str = "TLSv1",
    headers: list[tuple[str, str]] | None = None,
    use_colors: bool | None = None,
    app_dir: str | None = None,
    factory: bool = False,
    h11_max_incomplete_event_size: int | None = None,
) -> None:
    """
    Run an ASGI application.

    Args:
        app: ASGI application instance, factory function, or import string (e.g., "myapp:app")
        host: Host address to bind
        port: Port number to bind
        uds: Unix domain socket path (alternative to host/port)
        fd: File descriptor for socket binding (alternative to host/port)
        loop: Event loop implementation
        http: HTTP protocol implementation
        ws: WebSocket protocol implementation
        ws_max_size: Maximum WebSocket message size in bytes
        ws_max_queue: Maximum WebSocket message queue length
        ws_ping_interval: WebSocket ping interval in seconds
        ws_ping_timeout: WebSocket ping timeout in seconds
        ws_per_message_deflate: Enable WebSocket per-message deflate compression
        lifespan: Lifespan event handling mode
        interface: Application interface type (ASGI3, ASGI2, or WSGI)
        reload: Enable auto-reload on file changes
        reload_dirs: Directories to watch for changes (defaults to current directory)
        reload_includes: Glob patterns to include for reload watching
        reload_excludes: Glob patterns to exclude from reload watching
        reload_delay: Delay between checking for file changes in seconds
        workers: Number of worker processes (None for single process)
        env_file: Path to environment file to load
        log_config: Logging configuration (dict, file path, or file object)
        log_level: Logging level (string or int)
        access_log: Enable access logging
        proxy_headers: Enable parsing of X-Forwarded-* proxy headers
        server_header: Include Server header in responses
        date_header: Include Date header in responses
        forwarded_allow_ips: IPs to trust for proxy headers (comma-separated or list)
        root_path: ASGI root_path for mounted applications
        limit_concurrency: Maximum number of concurrent connections
        backlog: Socket listen backlog size
        limit_max_requests: Maximum requests before worker restart
        timeout_keep_alive: Keep-alive timeout in seconds
        timeout_graceful_shutdown: Graceful shutdown timeout in seconds
        timeout_worker_healthcheck: Worker health check timeout in seconds
        ssl_keyfile: SSL private key file path
        ssl_certfile: SSL certificate file path
        ssl_keyfile_password: Password for SSL private key
        ssl_version: SSL protocol version
        ssl_cert_reqs: SSL certificate requirements
        ssl_ca_certs: SSL CA certificates file path
        ssl_ciphers: SSL cipher configuration
        headers: Custom default headers to include in all responses
        use_colors: Enable colored log output (None for auto-detection)
        app_dir: Directory to add to sys.path before importing app (used by run(), not passed to Config)
        factory: Treat app as a factory function
        h11_max_incomplete_event_size: Maximum buffer size for h11 incomplete events
    """

class Server:
    """
    ASGI server implementation.

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

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

        Args:
            config: Server configuration object
        """

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

        Args:
            sockets: Pre-bound sockets (optional)
        """

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

        Args:
            sockets: Pre-bound sockets (optional)
        """

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

        Args:
            sockets: Pre-bound sockets (optional)
        """

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

        Args:
            sockets: Sockets to close (optional)
        """

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

    Attributes:
        total_requests: Total number of requests processed
        connections: Set of active protocol connections
        tasks: Set of active background tasks
        default_headers: Default headers to include in responses
    """

    total_requests: int
    connections: set[asyncio.Protocol]
    tasks: set[asyncio.Task[None]]
    default_headers: list[tuple[bytes, bytes]]

class Config:
    """
    Server configuration.

    All parameters from run() function are available as constructor parameters.
    """

    def __init__(
        self,
        app: ASGIApplication | Callable | str,
        host: str = "127.0.0.1",
        port: int = 8000,
        # ... all other parameters same as run() function
    ) -> None:
        """Initialize configuration with server settings."""

    @property
    def asgi_version(self) -> Literal["2.0", "3.0"]:
        """Get ASGI version based on interface."""

    @property
    def is_ssl(self) -> bool:
        """Check if SSL is configured."""

    @property
    def use_subprocess(self) -> bool:
        """Check if subprocess mode is used."""

    @property
    def should_reload(self) -> bool:
        """Check if auto-reload should be enabled."""

    def configure_logging(self) -> None:
        """Configure the logging system."""

    def load(self) -> None:
        """Load application and configure all settings."""

    def bind_socket(self) -> socket.socket:
        """Bind and return a socket."""

class ColourizedFormatter(logging.Formatter):
    """
    Custom log formatter with colored output support.

    Args:
        fmt: Log format string
        datefmt: Date format string
        style: Format style ('%', '{', or '$')
        use_colors: Enable colored output (None for auto-detection)
    """

    def color_level_name(self, level_name: str, level_no: int) -> str:
        """Colorize log level name."""

    def should_use_colors(self) -> bool:
        """Check if colors should be used."""

    def formatMessage(self, record: logging.LogRecord) -> str:
        """Format log record with colors."""

class AccessFormatter(ColourizedFormatter):
    """
    Formatter for HTTP access logs.

    Includes status code coloring and HTTP status phrases.
    """

    def get_status_code(self, status_code: int) -> str:
        """Format status code with HTTP phrase."""

class ProxyHeadersMiddleware:
    """
    Middleware for handling X-Forwarded-Proto and X-Forwarded-For headers.

    Args:
        app: ASGI application
        trusted_hosts: Trusted proxy hosts/networks (comma-separated or list)
    """

    def __init__(
        self,
        app: ASGI3Application,
        trusted_hosts: list[str] | str = "127.0.0.1",
    ) -> None: ...

    async def __call__(
        self,
        scope: Scope,
        receive: ASGIReceiveCallable,
        send: ASGISendCallable,
    ) -> None: ...

class ASGI2Middleware:
    """
    Adapter to run ASGI2 applications as ASGI3.

    Args:
        app: ASGI2 application class
    """

    def __init__(self, app: ASGI2Application) -> None: ...

    async def __call__(
        self,
        scope: Scope,
        receive: ASGIReceiveCallable,
        send: ASGISendCallable,
    ) -> None: ...

class WSGIMiddleware:
    """
    Adapter to run WSGI applications in ASGI.

    Args:
        app: WSGI application
        workers: Number of worker threads for blocking WSGI calls
    """

    def __init__(self, app: WSGIApp, workers: int = 10) -> None: ...

    async def __call__(
        self,
        scope: Scope,
        receive: ASGIReceiveCallable,
        send: ASGISendCallable,
    ) -> None: ...

# Application types
ASGIApplication = Union[ASGI2Application, ASGI3Application]
ASGI2Application = type[ASGI2Protocol]
ASGI3Application = Callable[[Scope, ASGIReceiveCallable, ASGISendCallable], Awaitable[None]]

# Scope types
Scope = Union[HTTPScope, WebSocketScope, LifespanScope]
WWWScope = Union[HTTPScope, WebSocketScope]

# Callable types
ASGIReceiveCallable = Callable[[], Awaitable[ASGIReceiveEvent]]
ASGISendCallable = Callable[[ASGISendEvent], Awaitable[None]]

class Multiprocess:
    """
    Supervisor for managing multiple worker processes.

    Args:
        config: Server configuration
        target: Target function to run in each worker
        sockets: Pre-bound sockets to share across workers
    """

    def __init__(
        self,
        config: Config,
        target: Callable,
        sockets: list[socket.socket],
    ) -> None: ...

    def run(self) -> None:
        """Run the multiprocess supervisor."""

class BaseReload:
    """
    Base class for reload supervisors.

    Monitors files for changes and restarts the worker process.

    Args:
        config: Server configuration
        target: Target function to run
        sockets: Pre-bound sockets
    """

    def run(self) -> None:
        """Run the reload supervisor."""

uvicorn [OPTIONS] APP

def main() -> None:
    """
    Main CLI entry point.

    Parses command-line arguments and runs the server.
    """

# Exit codes
STARTUP_FAILURE: int = 3

# Log levels
TRACE_LOG_LEVEL: int = 5

LOG_LEVELS: dict[str, int] = {
    "critical": logging.CRITICAL,
    "error": logging.ERROR,
    "warning": logging.WARNING,
    "info": logging.INFO,
    "debug": logging.DEBUG,
    "trace": TRACE_LOG_LEVEL,
}

# Protocol mappings
HTTP_PROTOCOLS: dict[str, str] = {
    "auto": "uvicorn.protocols.http.auto:AutoHTTPProtocol",
    "h11": "uvicorn.protocols.http.h11_impl:H11Protocol",
    "httptools": "uvicorn.protocols.http.httptools_impl:HttpToolsProtocol",
}

WS_PROTOCOLS: dict[str, str | None] = {
    "auto": "uvicorn.protocols.websockets.auto:AutoWebSocketsProtocol",
    "none": None,
    "websockets": "uvicorn.protocols.websockets.websockets_impl:WebSocketProtocol",
    "websockets-sansio": "uvicorn.protocols.websockets.websockets_sansio_impl:WebSocketsSansIOProtocol",
    "wsproto": "uvicorn.protocols.websockets.wsproto_impl:WSProtocol",
}

LIFESPAN: dict[str, str] = {
    "auto": "uvicorn.lifespan.on:LifespanOn",
    "on": "uvicorn.lifespan.on:LifespanOn",
    "off": "uvicorn.lifespan.off:LifespanOff",
}

LOOP_FACTORIES: dict[str, str | None] = {
    "none": None,
    "auto": "uvicorn.loops.auto:auto_loop_factory",
    "asyncio": "uvicorn.loops.asyncio:asyncio_loop_factory",
    "uvloop": "uvicorn.loops.uvloop:uvloop_loop_factory",
}

__version__: str = "0.37.0"