Загрузить файлы в «venv/Lib/site-packages/uvicorn»

venv/Lib/site-packages/uvicorn/_types.py

@@ -0,0 +1,274 @@
+"""
+Copyright (c) Django Software Foundation and individual contributors.
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+    1. Redistributions of source code must retain the above copyright notice,
+       this list of conditions and the following disclaimer.
+
+    2. Redistributions in binary form must reproduce the above copyright
+       notice, this list of conditions and the following disclaimer in the
+       documentation and/or other materials provided with the distribution.
+
+    3. Neither the name of Django nor the names of its contributors may be used
+       to endorse or promote products derived from this software without
+       specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+"""
+
+from __future__ import annotations
+
+import sys
+import types
+from collections.abc import Awaitable, Callable, Iterable, MutableMapping
+from typing import Any, Literal, Protocol, TypedDict
+
+if sys.version_info >= (3, 11):  # pragma: py-lt-311
+    from typing import NotRequired
+else:  # pragma: py-gte-311
+    from typing_extensions import NotRequired
+
+# WSGI
+Environ = MutableMapping[str, Any]
+ExcInfo = tuple[type[BaseException], BaseException, types.TracebackType | None]
+StartResponse = Callable[[str, Iterable[tuple[str, str]], ExcInfo | None], None]
+WSGIApp = Callable[[Environ, StartResponse], Iterable[bytes] | BaseException]
+
+
+# ASGI
+class ASGIVersions(TypedDict):
+    spec_version: str
+    version: Literal["2.0"] | Literal["3.0"]
+
+
+class HTTPScope(TypedDict):
+    type: Literal["http"]
+    asgi: ASGIVersions
+    http_version: str
+    method: str
+    scheme: str
+    path: str
+    raw_path: bytes
+    query_string: bytes
+    root_path: str
+    headers: Iterable[tuple[bytes, bytes]]
+    client: tuple[str, int] | None
+    server: tuple[str, int | None] | None
+    state: NotRequired[dict[str, Any]]
+    extensions: NotRequired[dict[str, dict[object, object]]]
+
+
+class WebSocketScope(TypedDict):
+    type: Literal["websocket"]
+    asgi: ASGIVersions
+    http_version: str
+    scheme: str
+    path: str
+    raw_path: bytes
+    query_string: bytes
+    root_path: str
+    headers: Iterable[tuple[bytes, bytes]]
+    client: tuple[str, int] | None
+    server: tuple[str, int | None] | None
+    subprotocols: Iterable[str]
+    state: NotRequired[dict[str, Any]]
+    extensions: NotRequired[dict[str, dict[object, object]]]
+
+
+class LifespanScope(TypedDict):
+    type: Literal["lifespan"]
+    asgi: ASGIVersions
+    state: NotRequired[dict[str, Any]]
+
+
+WWWScope = HTTPScope | WebSocketScope
+Scope = HTTPScope | WebSocketScope | LifespanScope
+
+
+class HTTPRequestEvent(TypedDict):
+    type: Literal["http.request"]
+    body: bytes
+    more_body: bool
+
+
+class HTTPResponseDebugEvent(TypedDict):
+    type: Literal["http.response.debug"]
+    info: dict[str, object]
+
+
+class HTTPResponseStartEvent(TypedDict):
+    type: Literal["http.response.start"]
+    status: int
+    headers: NotRequired[Iterable[tuple[bytes, bytes]]]
+    trailers: NotRequired[bool]
+
+
+class HTTPResponseBodyEvent(TypedDict):
+    type: Literal["http.response.body"]
+    body: bytes
+    more_body: NotRequired[bool]
+
+
+class HTTPResponseTrailersEvent(TypedDict):
+    type: Literal["http.response.trailers"]
+    headers: Iterable[tuple[bytes, bytes]]
+    more_trailers: bool
+
+
+class HTTPServerPushEvent(TypedDict):
+    type: Literal["http.response.push"]
+    path: str
+    headers: Iterable[tuple[bytes, bytes]]
+
+
+class HTTPDisconnectEvent(TypedDict):
+    type: Literal["http.disconnect"]
+
+
+class WebSocketConnectEvent(TypedDict):
+    type: Literal["websocket.connect"]
+
+
+class WebSocketAcceptEvent(TypedDict):
+    type: Literal["websocket.accept"]
+    subprotocol: NotRequired[str | None]
+    headers: NotRequired[Iterable[tuple[bytes, bytes]]]
+
+
+class _WebSocketReceiveEventBytes(TypedDict):
+    type: Literal["websocket.receive"]
+    bytes: bytes
+    text: NotRequired[None]
+
+
+class _WebSocketReceiveEventText(TypedDict):
+    type: Literal["websocket.receive"]
+    bytes: NotRequired[None]
+    text: str
+
+
+WebSocketReceiveEvent = _WebSocketReceiveEventBytes | _WebSocketReceiveEventText
+
+
+class _WebSocketSendEventBytes(TypedDict):
+    type: Literal["websocket.send"]
+    bytes: bytes
+    text: NotRequired[None]
+
+
+class _WebSocketSendEventText(TypedDict):
+    type: Literal["websocket.send"]
+    bytes: NotRequired[None]
+    text: str
+
+
+WebSocketSendEvent = _WebSocketSendEventBytes | _WebSocketSendEventText
+
+
+class WebSocketResponseStartEvent(TypedDict):
+    type: Literal["websocket.http.response.start"]
+    status: int
+    headers: Iterable[tuple[bytes, bytes]]
+
+
+class WebSocketResponseBodyEvent(TypedDict):
+    type: Literal["websocket.http.response.body"]
+    body: bytes
+    more_body: NotRequired[bool]
+
+
+class WebSocketDisconnectEvent(TypedDict):
+    type: Literal["websocket.disconnect"]
+    code: int
+    reason: NotRequired[str | None]
+
+
+class WebSocketCloseEvent(TypedDict):
+    type: Literal["websocket.close"]
+    code: NotRequired[int]
+    reason: NotRequired[str | None]
+
+
+class LifespanStartupEvent(TypedDict):
+    type: Literal["lifespan.startup"]
+
+
+class LifespanShutdownEvent(TypedDict):
+    type: Literal["lifespan.shutdown"]
+
+
+class LifespanStartupCompleteEvent(TypedDict):
+    type: Literal["lifespan.startup.complete"]
+
+
+class LifespanStartupFailedEvent(TypedDict):
+    type: Literal["lifespan.startup.failed"]
+    message: str
+
+
+class LifespanShutdownCompleteEvent(TypedDict):
+    type: Literal["lifespan.shutdown.complete"]
+
+
+class LifespanShutdownFailedEvent(TypedDict):
+    type: Literal["lifespan.shutdown.failed"]
+    message: str
+
+
+WebSocketEvent = WebSocketReceiveEvent | WebSocketDisconnectEvent | WebSocketConnectEvent
+
+
+ASGIReceiveEvent = (
+    HTTPRequestEvent
+    | HTTPDisconnectEvent
+    | WebSocketConnectEvent
+    | WebSocketReceiveEvent
+    | WebSocketDisconnectEvent
+    | LifespanStartupEvent
+    | LifespanShutdownEvent
+)
+
+
+ASGISendEvent = (
+    HTTPResponseStartEvent
+    | HTTPResponseBodyEvent
+    | HTTPResponseTrailersEvent
+    | HTTPServerPushEvent
+    | HTTPDisconnectEvent
+    | WebSocketAcceptEvent
+    | WebSocketSendEvent
+    | WebSocketResponseStartEvent
+    | WebSocketResponseBodyEvent
+    | WebSocketCloseEvent
+    | LifespanStartupCompleteEvent
+    | LifespanStartupFailedEvent
+    | LifespanShutdownCompleteEvent
+    | LifespanShutdownFailedEvent
+)
+
+
+ASGIReceiveCallable = Callable[[], Awaitable[ASGIReceiveEvent]]
+ASGISendCallable = Callable[[ASGISendEvent], Awaitable[None]]
+
+
+class ASGI2Protocol(Protocol):
+    def __init__(self, scope: Scope) -> None: ...  # pragma: no cover
+
+    async def __call__(self, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: ...  # pragma: no cover
+
+
+ASGI2Application = type[ASGI2Protocol]
+ASGI3Application = Callable[[Scope, ASGIReceiveCallable, ASGISendCallable], Awaitable[None]]
+ASGIApplication = ASGI2Application | ASGI3Application

venv/Lib/site-packages/uvicorn/config.py

@@ -0,0 +1,588 @@
+from __future__ import annotations
+
+import asyncio
+import inspect
+import json
+import logging
+import logging.config
+import os
+import socket
+import ssl
+import sys
+from collections.abc import Awaitable, Callable
+from configparser import RawConfigParser
+from pathlib import Path
+from typing import IO, Any, Literal
+
+import click
+
+from uvicorn._compat import iscoroutinefunction
+from uvicorn._types import ASGIApplication
+from uvicorn.importer import ImportFromStringError, import_from_string
+from uvicorn.logging import TRACE_LOG_LEVEL
+from uvicorn.middleware.asgi2 import ASGI2Middleware
+from uvicorn.middleware.message_logger import MessageLoggerMiddleware
+from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
+from uvicorn.middleware.wsgi import WSGIMiddleware
+
+HTTPProtocolType = Literal["auto", "h11", "httptools"]
+WSProtocolType = Literal["auto", "none", "websockets", "websockets-sansio", "wsproto"]
+LifespanType = Literal["auto", "on", "off"]
+LoopFactoryType = Literal["none", "auto", "asyncio", "uvloop"]
+InterfaceType = Literal["auto", "asgi3", "asgi2", "wsgi"]
+
+LOG_LEVELS: dict[str, int] = {
+    "critical": logging.CRITICAL,
+    "error": logging.ERROR,
+    "warning": logging.WARNING,
+    "info": logging.INFO,
+    "debug": logging.DEBUG,
+    "trace": TRACE_LOG_LEVEL,
+}
+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",
+}
+INTERFACES: list[InterfaceType] = ["auto", "asgi3", "asgi2", "wsgi"]
+
+SSL_PROTOCOL_VERSION: int = ssl.PROTOCOL_TLS_SERVER
+
+LOGGING_CONFIG: dict[str, Any] = {
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {
+        "default": {
+            "()": "uvicorn.logging.DefaultFormatter",
+            "fmt": "%(levelprefix)s %(message)s",
+            "use_colors": None,
+        },
+        "access": {
+            "()": "uvicorn.logging.AccessFormatter",
+            "fmt": '%(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',  # noqa: E501
+        },
+    },
+    "handlers": {
+        "default": {
+            "formatter": "default",
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stderr",
+        },
+        "access": {
+            "formatter": "access",
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stdout",
+        },
+    },
+    "loggers": {
+        "uvicorn": {"handlers": ["default"], "level": "INFO", "propagate": False},
+        "uvicorn.error": {"level": "INFO"},
+        "uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
+    },
+}
+
+logger = logging.getLogger("uvicorn.error")
+
+
+def create_ssl_context(
+    certfile: str | os.PathLike[str],
+    keyfile: str | os.PathLike[str] | None,
+    password: str | None,
+    ssl_version: int,
+    cert_reqs: int,
+    ca_certs: str | os.PathLike[str] | None,
+    ciphers: str | None,
+) -> ssl.SSLContext:
+    ctx = ssl.SSLContext(ssl_version)
+    get_password = (lambda: password) if password else None
+    ctx.load_cert_chain(certfile, keyfile, get_password)
+    ctx.verify_mode = ssl.VerifyMode(cert_reqs)
+    if ca_certs:
+        ctx.load_verify_locations(ca_certs)
+    if ciphers:
+        ctx.set_ciphers(ciphers)
+    return ctx
+
+
+def is_dir(path: Path) -> bool:
+    try:
+        if not path.is_absolute():
+            path = path.resolve()
+        return path.is_dir()
+    except OSError:  # pragma: full coverage
+        return False
+
+
+def resolve_reload_patterns(patterns_list: list[str], directories_list: list[str]) -> tuple[list[str], list[Path]]:
+    directories: list[Path] = list(set(map(Path, directories_list.copy())))
+    patterns: list[str] = patterns_list.copy()
+
+    current_working_directory = Path.cwd()
+    for pattern in patterns_list:
+        # Special case for the .* pattern, otherwise this would only match
+        # hidden directories which is probably undesired
+        if pattern == ".*":
+            continue  # pragma: py-not-linux
+        patterns.append(pattern)
+        if is_dir(Path(pattern)):
+            directories.append(Path(pattern))
+        else:
+            for match in current_working_directory.glob(pattern):
+                if is_dir(match):
+                    directories.append(match)
+
+    directories = list(set(directories))
+    directories = list(map(Path, directories))
+    directories = list(map(lambda x: x.resolve(), directories))
+    directories = list({reload_path for reload_path in directories if is_dir(reload_path)})
+
+    children = []
+    for j in range(len(directories)):
+        for k in range(j + 1, len(directories)):  # pragma: full coverage
+            if directories[j] in directories[k].parents:
+                children.append(directories[k])
+            elif directories[k] in directories[j].parents:
+                children.append(directories[j])
+
+    directories = list(set(directories).difference(set(children)))
+
+    return list(set(patterns)), directories
+
+
+def _normalize_dirs(dirs: list[str] | str | None) -> list[str]:
+    if dirs is None:
+        return []
+    if isinstance(dirs, str):
+        return [dirs]
+    return list(set(dirs))
+
+
+class Config:
+    def __init__(
+        self,
+        app: ASGIApplication | Callable[..., Any] | str,
+        host: str = "127.0.0.1",
+        port: int = 8000,
+        uds: str | None = None,
+        fd: int | None = None,
+        loop: LoopFactoryType | str = "auto",
+        http: type[asyncio.Protocol] | HTTPProtocolType | str = "auto",
+        ws: type[asyncio.Protocol] | WSProtocolType | str = "auto",
+        ws_max_size: int = 16 * 1024 * 1024,
+        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: LifespanType = "auto",
+        env_file: str | os.PathLike[str] | None = None,
+        log_config: dict[str, Any] | str | os.PathLike[str] | RawConfigParser | IO[Any] | None = LOGGING_CONFIG,
+        log_level: str | int | None = None,
+        access_log: bool = True,
+        use_colors: bool | None = None,
+        interface: InterfaceType = "auto",
+        reload: bool = False,
+        reload_dirs: list[str] | str | None = None,
+        reload_delay: float = 0.25,
+        reload_includes: list[str] | str | None = None,
+        reload_excludes: list[str] | str | None = None,
+        workers: int | None = None,
+        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,
+        limit_max_requests: int | None = None,
+        limit_max_requests_jitter: int = 0,
+        backlog: int = 2048,
+        timeout_keep_alive: int = 5,
+        timeout_notify: int = 30,
+        timeout_graceful_shutdown: int | None = None,
+        timeout_worker_healthcheck: int = 5,
+        callback_notify: Callable[..., Awaitable[None]] | None = None,
+        ssl_keyfile: str | os.PathLike[str] | None = None,
+        ssl_certfile: str | os.PathLike[str] | None = None,
+        ssl_keyfile_password: str | None = None,
+        ssl_version: int = SSL_PROTOCOL_VERSION,
+        ssl_cert_reqs: int = ssl.CERT_NONE,
+        ssl_ca_certs: str | os.PathLike[str] | None = None,
+        ssl_ciphers: str | None = None,
+        ssl_context_factory: Callable[[Config, Callable[[], ssl.SSLContext]], ssl.SSLContext] | None = None,
+        headers: list[tuple[str, str]] | None = None,
+        factory: bool = False,
+        h11_max_incomplete_event_size: int | None = None,
+        reset_contextvars: bool = False,
+    ):
+        self.app = app
+        self.host = host
+        self.port = port
+        self.uds = uds
+        self.fd = fd
+        self.loop = loop
+        self.http = http
+        self.ws = ws
+        self.ws_max_size = ws_max_size
+        self.ws_max_queue = ws_max_queue
+        self.ws_ping_interval = ws_ping_interval
+        self.ws_ping_timeout = ws_ping_timeout
+        self.ws_per_message_deflate = ws_per_message_deflate
+        self.lifespan = lifespan
+        self.log_config = log_config
+        self.log_level = log_level
+        self.access_log = access_log
+        self.use_colors = use_colors
+        self.interface = interface
+        self.reload = reload
+        self.reload_delay = reload_delay
+        self.workers = workers or 1
+        self.proxy_headers = proxy_headers
+        self.server_header = server_header
+        self.date_header = date_header
+        self.root_path = root_path
+        self.limit_concurrency = limit_concurrency
+        self.limit_max_requests = limit_max_requests
+        self.limit_max_requests_jitter = limit_max_requests_jitter
+        self.backlog = backlog
+        self.timeout_keep_alive = timeout_keep_alive
+        self.timeout_notify = timeout_notify
+        self.timeout_graceful_shutdown = timeout_graceful_shutdown
+        self.timeout_worker_healthcheck = timeout_worker_healthcheck
+        self.callback_notify = callback_notify
+        self.ssl_keyfile = ssl_keyfile
+        self.ssl_certfile = ssl_certfile
+        self.ssl_keyfile_password = ssl_keyfile_password
+        self.ssl_version = ssl_version
+        self.ssl_cert_reqs = ssl_cert_reqs
+        self.ssl_ca_certs = ssl_ca_certs
+        self.ssl_ciphers = ssl_ciphers
+        self.ssl_context_factory = ssl_context_factory
+        self.headers: list[tuple[str, str]] = headers or []
+        self.encoded_headers: list[tuple[bytes, bytes]] = []
+        self.factory = factory
+        self.h11_max_incomplete_event_size = h11_max_incomplete_event_size
+        self.reset_contextvars = reset_contextvars
+
+        self.loaded = False
+        self.configure_logging()
+
+        self.reload_dirs: list[Path] = []
+        self.reload_dirs_excludes: list[Path] = []
+        self.reload_includes: list[str] = []
+        self.reload_excludes: list[str] = []
+
+        if (reload_dirs or reload_includes or reload_excludes) and not self.should_reload:
+            logger.warning(
+                "Current configuration will not reload as not all conditions are met, please refer to documentation."
+            )
+
+        if self.should_reload:
+            reload_dirs = _normalize_dirs(reload_dirs)
+            reload_includes = _normalize_dirs(reload_includes)
+            reload_excludes = _normalize_dirs(reload_excludes)
+
+            self.reload_includes, self.reload_dirs = resolve_reload_patterns(reload_includes, reload_dirs)
+
+            self.reload_excludes, self.reload_dirs_excludes = resolve_reload_patterns(reload_excludes, [])
+
+            reload_dirs_tmp = self.reload_dirs.copy()
+
+            for directory in self.reload_dirs_excludes:
+                for reload_directory in reload_dirs_tmp:
+                    if directory == reload_directory or directory in reload_directory.parents:
+                        try:
+                            self.reload_dirs.remove(reload_directory)
+                        except ValueError:  # pragma: full coverage
+                            pass
+
+            for pattern in self.reload_excludes:
+                if pattern in self.reload_includes:
+                    self.reload_includes.remove(pattern)  # pragma: full coverage
+
+            if not self.reload_dirs:
+                if reload_dirs:
+                    logger.warning(
+                        "Provided reload directories %s did not contain valid "
+                        + "directories, watching current working directory.",
+                        reload_dirs,
+                    )
+                self.reload_dirs = [Path.cwd()]
+
+            logger.info(
+                "Will watch for changes in these directories: %s",
+                sorted(list(map(str, self.reload_dirs))),
+            )
+
+        if env_file is not None:
+            from dotenv import load_dotenv
+
+            logger.info("Loading environment from '%s'", env_file)
+            load_dotenv(dotenv_path=env_file)
+
+        if workers is None and "WEB_CONCURRENCY" in os.environ:
+            self.workers = int(os.environ["WEB_CONCURRENCY"])
+
+        self.forwarded_allow_ips: list[str] | str
+        if forwarded_allow_ips is None:
+            self.forwarded_allow_ips = os.environ.get("FORWARDED_ALLOW_IPS", "127.0.0.1")
+        else:
+            self.forwarded_allow_ips = forwarded_allow_ips  # pragma: full coverage
+
+        if self.reload and self.workers > 1:
+            logger.warning('"workers" flag is ignored when reloading is enabled.')
+
+    @property
+    def asgi_version(self) -> Literal["2.0", "3.0"]:
+        mapping: dict[str, Literal["2.0", "3.0"]] = {
+            "asgi2": "2.0",
+            "asgi3": "3.0",
+            "wsgi": "3.0",
+        }
+        return mapping[self.interface]
+
+    @property
+    def is_ssl(self) -> bool:
+        return bool(self.ssl_keyfile or self.ssl_certfile or self.ssl_context_factory)
+
+    @property
+    def use_subprocess(self) -> bool:
+        return bool(self.reload or self.workers > 1)
+
+    def configure_logging(self) -> None:
+        logging.addLevelName(TRACE_LOG_LEVEL, "TRACE")
+
+        if self.log_config is not None:
+            if isinstance(self.log_config, os.PathLike):
+                self.log_config = os.fspath(self.log_config)
+
+            if isinstance(self.log_config, dict):
+                if self.use_colors in (True, False):
+                    self.log_config["formatters"]["default"]["use_colors"] = self.use_colors
+                    self.log_config["formatters"]["access"]["use_colors"] = self.use_colors
+                logging.config.dictConfig(self.log_config)
+            elif isinstance(self.log_config, str) and self.log_config.endswith(".json"):
+                with open(self.log_config) as file:
+                    loaded_config = json.load(file)
+                    logging.config.dictConfig(loaded_config)
+            elif isinstance(self.log_config, str) and self.log_config.endswith((".yaml", ".yml")):
+                try:
+                    import yaml
+                except ImportError as e:
+                    raise ImportError(
+                        "Install the PyYAML package or uvicorn[standard] to use `--log-config` with YAML files."
+                    ) from e
+
+                with open(self.log_config) as file:
+                    loaded_config = yaml.safe_load(file)
+                    logging.config.dictConfig(loaded_config)
+            else:
+                # See the note about fileConfig() here:
+                # https://docs.python.org/3/library/logging.config.html#configuration-file-format
+                logging.config.fileConfig(self.log_config, disable_existing_loggers=False)
+
+        if self.log_level is not None:
+            if isinstance(self.log_level, str):
+                log_level = LOG_LEVELS[self.log_level.lower()]
+            else:
+                log_level = self.log_level
+            logging.getLogger("uvicorn.error").setLevel(log_level)
+            logging.getLogger("uvicorn.access").setLevel(log_level)
+            logging.getLogger("uvicorn.asgi").setLevel(log_level)
+        if self.access_log is False:
+            logging.getLogger("uvicorn.access").handlers = []
+            logging.getLogger("uvicorn.access").propagate = False
+
+    def load_app(self) -> Any:
+        """Import the app and return it. Exits on failure."""
+        try:
+            return import_from_string(self.app)
+        except ImportFromStringError as exc:
+            logger.error("Error loading ASGI app. %s" % exc)
+            sys.exit(1)
+
+    def load(self) -> None:
+        assert not self.loaded
+
+        if self.ssl_context_factory is not None:
+
+            def default_factory() -> ssl.SSLContext:
+                if not self.ssl_certfile:
+                    raise RuntimeError(
+                        "`default_ssl_context_factory()` requires `ssl_certfile` to be set on `Config`. "
+                        "Either pass `ssl_certfile` (and optionally `ssl_keyfile`) or build the `SSLContext` "
+                        "directly inside `ssl_context_factory` without calling the default factory."
+                    )
+                return create_ssl_context(
+                    keyfile=self.ssl_keyfile,
+                    certfile=self.ssl_certfile,
+                    password=self.ssl_keyfile_password,
+                    ssl_version=self.ssl_version,
+                    cert_reqs=self.ssl_cert_reqs,
+                    ca_certs=self.ssl_ca_certs,
+                    ciphers=self.ssl_ciphers,
+                )
+
+            context = self.ssl_context_factory(self, default_factory)
+            if not isinstance(context, ssl.SSLContext):
+                raise TypeError(f"`ssl_context_factory` must return an `ssl.SSLContext`, got {type(context).__name__}")
+            self.ssl: ssl.SSLContext | None = context
+        elif self.is_ssl:
+            assert self.ssl_certfile
+            self.ssl = create_ssl_context(
+                keyfile=self.ssl_keyfile,
+                certfile=self.ssl_certfile,
+                password=self.ssl_keyfile_password,
+                ssl_version=self.ssl_version,
+                cert_reqs=self.ssl_cert_reqs,
+                ca_certs=self.ssl_ca_certs,
+                ciphers=self.ssl_ciphers,
+            )
+        else:
+            self.ssl = None
+
+        encoded_headers = [(key.lower().encode("latin1"), value.encode("latin1")) for key, value in self.headers]
+        self.encoded_headers = (
+            [(b"server", b"uvicorn")] + encoded_headers
+            if b"server" not in dict(encoded_headers) and self.server_header
+            else encoded_headers
+        )
+
+        if isinstance(self.http, str):
+            http_protocol_class = import_from_string(HTTP_PROTOCOLS.get(self.http, self.http))
+            self.http_protocol_class: type[asyncio.Protocol] = http_protocol_class
+        else:
+            self.http_protocol_class = self.http
+
+        if isinstance(self.ws, str):
+            ws_protocol_class = import_from_string(WS_PROTOCOLS.get(self.ws, self.ws))
+            self.ws_protocol_class: type[asyncio.Protocol] | None = ws_protocol_class
+        else:
+            self.ws_protocol_class = self.ws
+
+        self.lifespan_class = import_from_string(LIFESPAN[self.lifespan])
+
+        self.loaded_app = self.load_app()
+
+        try:
+            self.loaded_app = self.loaded_app()
+        except TypeError as exc:
+            if self.factory:
+                logger.error("Error loading ASGI app factory: %s", exc)
+                sys.exit(1)
+        else:
+            if not self.factory:
+                logger.warning(
+                    "ASGI app factory detected. Using it, but please consider setting the --factory flag explicitly."
+                )
+
+        if self.interface == "auto":
+            if inspect.isclass(self.loaded_app):
+                use_asgi_3 = hasattr(self.loaded_app, "__await__")
+            elif inspect.isfunction(self.loaded_app):
+                use_asgi_3 = iscoroutinefunction(self.loaded_app)
+            else:
+                call = getattr(self.loaded_app, "__call__", None)
+                use_asgi_3 = iscoroutinefunction(call)
+            self.interface = "asgi3" if use_asgi_3 else "asgi2"
+
+        if self.interface == "wsgi":
+            self.loaded_app = WSGIMiddleware(self.loaded_app)
+            self.ws_protocol_class = None
+        elif self.interface == "asgi2":
+            self.loaded_app = ASGI2Middleware(self.loaded_app)
+
+        if logger.getEffectiveLevel() <= TRACE_LOG_LEVEL:
+            self.loaded_app = MessageLoggerMiddleware(self.loaded_app)
+        if self.proxy_headers:
+            self.loaded_app = ProxyHeadersMiddleware(self.loaded_app, trusted_hosts=self.forwarded_allow_ips)
+
+        self.loaded = True
+
+    def setup_event_loop(self) -> None:
+        raise AttributeError(
+            "The `setup_event_loop` method was replaced by `get_loop_factory` in uvicorn 0.36.0.\n"
+            "None of those methods are supposed to be used directly. If you are doing it, please let me know here: "
+            "https://github.com/Kludex/uvicorn/discussions/2706. Thank you, and sorry for the inconvenience."
+        )
+
+    def get_loop_factory(self) -> Callable[[], asyncio.AbstractEventLoop] | None:
+        if self.loop in LOOP_FACTORIES:
+            loop_factory: Callable[..., Any] | None = import_from_string(LOOP_FACTORIES[self.loop])
+        else:
+            try:
+                return import_from_string(self.loop)
+            except ImportFromStringError as exc:
+                logger.error("Error loading custom loop setup function. %s" % exc)
+                sys.exit(1)
+        if loop_factory is None:
+            return None
+        return loop_factory(use_subprocess=self.use_subprocess)
+
+    def bind_socket(self) -> socket.socket:
+        logger_args: list[str | int]
+        if self.uds is not None:  # pragma: py-win32
+            path = self.uds
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            try:
+                sock.bind(path)
+                uds_perms = 0o666
+                os.chmod(self.uds, uds_perms)
+            except OSError as exc:  # pragma: full coverage
+                logger.error(exc)
+                sys.exit(1)
+
+            message = "Uvicorn running on unix socket %s (Press CTRL+C to quit)"
+            sock_name_format = "%s"
+            color_message = "Uvicorn running on " + click.style(sock_name_format, bold=True) + " (Press CTRL+C to quit)"
+            logger_args = [self.uds]
+        elif self.fd is not None:  # pragma: py-win32
+            sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
+            message = "Uvicorn running on socket %s (Press CTRL+C to quit)"
+            fd_name_format = "%s"
+            color_message = "Uvicorn running on " + click.style(fd_name_format, bold=True) + " (Press CTRL+C to quit)"
+            logger_args = [sock.getsockname()]
+        else:
+            family = socket.AF_INET
+            addr_format = "%s://%s:%d"
+
+            if self.host and ":" in self.host:  # pragma: full coverage
+                # It's an IPv6 address.
+                family = socket.AF_INET6
+                addr_format = "%s://[%s]:%d"
+
+            sock = socket.socket(family=family)
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            try:
+                sock.bind((self.host, self.port))
+            except OSError as exc:  # pragma: full coverage
+                logger.error(exc)
+                sys.exit(1)
+
+            message = f"Uvicorn running on {addr_format} (Press CTRL+C to quit)"
+            color_message = "Uvicorn running on " + click.style(addr_format, bold=True) + " (Press CTRL+C to quit)"
+            protocol_name = "https" if self.is_ssl else "http"
+            logger_args = [protocol_name, self.host, sock.getsockname()[1]]
+        logger.info(message, *logger_args, extra={"color_message": color_message})
+        sock.set_inheritable(True)
+        return sock
+
+    @property
+    def should_reload(self) -> bool:
+        return isinstance(self.app, str) and self.reload

venv/Lib/site-packages/uvicorn/importer.py

@@ -0,0 +1,34 @@
+import importlib
+from typing import Any
+
+
+class ImportFromStringError(Exception):
+    pass
+
+
+def import_from_string(import_str: Any) -> Any:
+    if not isinstance(import_str, str):
+        return import_str
+
+    module_str, _, attrs_str = import_str.partition(":")
+    if not module_str or not attrs_str:
+        message = 'Import string "{import_str}" must be in format "<module>:<attribute>".'
+        raise ImportFromStringError(message.format(import_str=import_str))
+
+    try:
+        module = importlib.import_module(module_str)
+    except ModuleNotFoundError as exc:
+        if exc.name != module_str:
+            raise exc from None
+        message = 'Could not import module "{module_str}".'
+        raise ImportFromStringError(message.format(module_str=module_str))
+
+    instance = module
+    try:
+        for attr_str in attrs_str.split("."):
+            instance = getattr(instance, attr_str)
+    except AttributeError:
+        message = 'Attribute "{attrs_str}" not found in module "{module_str}".'
+        raise ImportFromStringError(message.format(attrs_str=attrs_str, module_str=module_str))
+
+    return instance

venv/Lib/site-packages/uvicorn/logging.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import http
+import logging
+import sys
+from copy import copy
+from typing import Literal
+
+import click
+
+TRACE_LOG_LEVEL = 5
+
+
+class ColourizedFormatter(logging.Formatter):
+    """
+    A custom log formatter class that:
+
+    * Outputs the LOG_LEVEL with an appropriate color.
+    * If a log call includes an `extra={"color_message": ...}` it will be used
+      for formatting the output, instead of the plain text message.
+    """
+
+    level_name_colors = {
+        TRACE_LOG_LEVEL: lambda level_name: click.style(str(level_name), fg="blue"),
+        logging.DEBUG: lambda level_name: click.style(str(level_name), fg="cyan"),
+        logging.INFO: lambda level_name: click.style(str(level_name), fg="green"),
+        logging.WARNING: lambda level_name: click.style(str(level_name), fg="yellow"),
+        logging.ERROR: lambda level_name: click.style(str(level_name), fg="red"),
+        logging.CRITICAL: lambda level_name: click.style(str(level_name), fg="bright_red"),
+    }
+
+    def __init__(
+        self,
+        fmt: str | None = None,
+        datefmt: str | None = None,
+        style: Literal["%", "{", "$"] = "%",
+        use_colors: bool | None = None,
+    ):
+        if use_colors in (True, False):
+            self.use_colors = use_colors
+        else:
+            self.use_colors = sys.stdout.isatty()
+        super().__init__(fmt=fmt, datefmt=datefmt, style=style)
+
+    def color_level_name(self, level_name: str, level_no: int) -> str:
+        def default(level_name: str) -> str:
+            return str(level_name)  # pragma: no cover
+
+        func = self.level_name_colors.get(level_no, default)
+        return func(level_name)
+
+    def should_use_colors(self) -> bool:
+        return True  # pragma: no cover
+
+    def formatMessage(self, record: logging.LogRecord) -> str:
+        recordcopy = copy(record)
+        levelname = recordcopy.levelname
+        separator = " " * (8 - len(recordcopy.levelname))
+        if self.use_colors:
+            levelname = self.color_level_name(levelname, recordcopy.levelno)
+            if "color_message" in recordcopy.__dict__:
+                recordcopy.msg = recordcopy.__dict__["color_message"]
+                recordcopy.__dict__["message"] = recordcopy.getMessage()
+        recordcopy.__dict__["levelprefix"] = levelname + ":" + separator
+        return super().formatMessage(recordcopy)
+
+
+class DefaultFormatter(ColourizedFormatter):
+    def should_use_colors(self) -> bool:
+        return sys.stderr.isatty()  # pragma: no cover
+
+
+class AccessFormatter(ColourizedFormatter):
+    status_code_colours = {
+        1: lambda code: click.style(str(code), fg="bright_white"),
+        2: lambda code: click.style(str(code), fg="green"),
+        3: lambda code: click.style(str(code), fg="yellow"),
+        4: lambda code: click.style(str(code), fg="red"),
+        5: lambda code: click.style(str(code), fg="bright_red"),
+    }
+
+    def get_status_code(self, status_code: int) -> str:
+        try:
+            status_phrase = http.HTTPStatus(status_code).phrase
+        except ValueError:
+            status_phrase = ""
+        status_and_phrase = f"{status_code} {status_phrase}"
+        if self.use_colors:
+
+            def default(code: int) -> str:
+                return status_and_phrase  # pragma: no cover
+
+            func = self.status_code_colours.get(status_code // 100, default)
+            return func(status_and_phrase)
+        return status_and_phrase
+
+    def formatMessage(self, record: logging.LogRecord) -> str:
+        recordcopy = copy(record)
+        (
+            client_addr,
+            method,
+            full_path,
+            http_version,
+            status_code,
+        ) = recordcopy.args  # type: ignore[misc]
+        status_code = self.get_status_code(int(status_code))  # type: ignore[arg-type]
+        request_line = f"{method} {full_path} HTTP/{http_version}"
+        if self.use_colors:
+            request_line = click.style(request_line, bold=True)
+        recordcopy.__dict__.update(
+            {
+                "client_addr": client_addr,
+                "request_line": request_line,
+                "status_code": status_code,
+            }
+        )
+        return super().formatMessage(recordcopy)

venv/Lib/site-packages/uvicorn/main.py

@@ -0,0 +1,644 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import platform
+import ssl
+import sys
+import warnings
+from collections.abc import Callable
+from configparser import RawConfigParser
+from typing import IO, Any, get_args
+
+import click
+
+import uvicorn
+from uvicorn._types import ASGIApplication
+from uvicorn.config import (
+    INTERFACES,
+    LIFESPAN,
+    LOG_LEVELS,
+    LOGGING_CONFIG,
+    SSL_PROTOCOL_VERSION,
+    Config,
+    HTTPProtocolType,
+    InterfaceType,
+    LifespanType,
+    LoopFactoryType,
+    WSProtocolType,
+)
+from uvicorn.server import Server
+from uvicorn.supervisors import ChangeReload, Multiprocess
+
+LEVEL_CHOICES = click.Choice(list(LOG_LEVELS.keys()))
+LIFESPAN_CHOICES = click.Choice(list(LIFESPAN.keys()))
+INTERFACE_CHOICES = click.Choice(INTERFACES)
+
+
+def _metavar_from_type(_type: Any) -> str:
+    return f"[{'|'.join(key for key in get_args(_type) if key != 'none')}]"
+
+
+STARTUP_FAILURE = 3
+
+logger = logging.getLogger("uvicorn.error")
+
+
+def print_version(ctx: click.Context, param: click.Parameter, value: bool) -> None:
+    if not value or ctx.resilient_parsing:
+        return
+    click.echo(
+        "Running uvicorn {version} with {py_implementation} {py_version} on {system}".format(  # noqa: UP032
+            version=uvicorn.__version__,
+            py_implementation=platform.python_implementation(),
+            py_version=platform.python_version(),
+            system=platform.system(),
+        )
+    )
+    ctx.exit()
+
+
+@click.command(context_settings={"auto_envvar_prefix": "UVICORN"})
+@click.argument("app", envvar="UVICORN_APP")
+@click.option(
+    "--host",
+    type=str,
+    default="127.0.0.1",
+    help="Bind socket to this host.",
+    show_default=True,
+)
+@click.option(
+    "--port",
+    type=int,
+    default=8000,
+    help="Bind socket to this port. If 0, an available port will be picked.",
+    show_default=True,
+)
+@click.option("--uds", type=str, default=None, help="Bind to a UNIX domain socket.")
+@click.option("--fd", type=int, default=None, help="Bind to socket from this file descriptor.")
+@click.option("--reload", is_flag=True, default=False, help="Enable auto-reload.")
+@click.option(
+    "--reload-dir",
+    "reload_dirs",
+    multiple=True,
+    help="Set reload directories explicitly, instead of using the current working directory.",
+    type=click.Path(exists=True),
+)
+@click.option(
+    "--reload-include",
+    "reload_includes",
+    multiple=True,
+    help="Set glob patterns to include while watching for files. Includes '*.py' "
+    "by default; these defaults can be overridden with `--reload-exclude`. "
+    "This option has no effect unless watchfiles is installed.",
+)
+@click.option(
+    "--reload-exclude",
+    "reload_excludes",
+    multiple=True,
+    help="Set glob patterns to exclude while watching for files. Includes "
+    "'.*, .py[cod], .sw.*, ~*' by default; these defaults can be overridden "
+    "with `--reload-include`. This option has no effect unless watchfiles is "
+    "installed.",
+)
+@click.option(
+    "--reload-delay",
+    type=float,
+    default=0.25,
+    show_default=True,
+    help="Delay between previous and next check if application needs to be. Defaults to 0.25s.",
+)
+@click.option(
+    "--workers",
+    default=None,
+    type=int,
+    help="Number of worker processes. Defaults to the $WEB_CONCURRENCY environment"
+    " variable if available, or 1. Not valid with --reload.",
+)
+@click.option(
+    "--loop",
+    type=str,
+    metavar=_metavar_from_type(LoopFactoryType),
+    default="auto",
+    help="Event loop factory implementation.",
+    show_default=True,
+)
+@click.option(
+    "--http",
+    type=str,
+    metavar=_metavar_from_type(HTTPProtocolType),
+    default="auto",
+    help="HTTP protocol implementation.",
+    show_default=True,
+)
+@click.option(
+    "--ws",
+    type=str,
+    metavar=_metavar_from_type(WSProtocolType),
+    default="auto",
+    help="WebSocket protocol implementation.",
+    show_default=True,
+)
+@click.option(
+    "--ws-max-size",
+    type=int,
+    default=16777216,
+    help="WebSocket max size message in bytes",
+    show_default=True,
+)
+@click.option(
+    "--ws-max-queue",
+    type=int,
+    default=32,
+    help="The maximum length of the WebSocket message queue.",
+    show_default=True,
+)
+@click.option(
+    "--ws-ping-interval",
+    type=float,
+    default=20.0,
+    help="WebSocket ping interval in seconds.",
+    show_default=True,
+)
+@click.option(
+    "--ws-ping-timeout",
+    type=float,
+    default=20.0,
+    help="WebSocket ping timeout in seconds.",
+    show_default=True,
+)
+@click.option(
+    "--ws-per-message-deflate",
+    type=bool,
+    default=True,
+    help="WebSocket per-message-deflate compression",
+    show_default=True,
+)
+@click.option(
+    "--lifespan",
+    type=LIFESPAN_CHOICES,
+    default="auto",
+    help="Lifespan implementation.",
+    show_default=True,
+)
+@click.option(
+    "--interface",
+    type=INTERFACE_CHOICES,
+    default="auto",
+    help="Select ASGI3, ASGI2, or WSGI as the application interface.",
+    show_default=True,
+)
+@click.option(
+    "--env-file",
+    type=click.Path(exists=True),
+    default=None,
+    help="Environment configuration file.",
+    show_default=True,
+)
+@click.option(
+    "--log-config",
+    type=click.Path(exists=True),
+    default=None,
+    help="Logging configuration file. Supported formats: .ini, .json, .yaml.",
+    show_default=True,
+)
+@click.option(
+    "--log-level",
+    type=LEVEL_CHOICES,
+    default=None,
+    help="Log level. [default: info]",
+    show_default=True,
+)
+@click.option(
+    "--access-log/--no-access-log",
+    is_flag=True,
+    default=True,
+    help="Enable/Disable access log.",
+)
+@click.option(
+    "--use-colors/--no-use-colors",
+    is_flag=True,
+    default=None,
+    help="Enable/Disable colorized logging.",
+)
+@click.option(
+    "--proxy-headers/--no-proxy-headers",
+    is_flag=True,
+    default=True,
+    help="Enable/Disable X-Forwarded-Proto, X-Forwarded-For to populate url scheme and remote address info.",
+)
+@click.option(
+    "--server-header/--no-server-header",
+    is_flag=True,
+    default=True,
+    help="Enable/Disable default Server header.",
+)
+@click.option(
+    "--date-header/--no-date-header",
+    is_flag=True,
+    default=True,
+    help="Enable/Disable default Date header.",
+)
+@click.option(
+    "--forwarded-allow-ips",
+    type=str,
+    default=None,
+    help="Comma separated list of IP Addresses, IP Networks, or literals "
+    "(e.g. UNIX Socket path) to trust with proxy headers. Defaults to the "
+    "$FORWARDED_ALLOW_IPS environment variable if available, or '127.0.0.1'. "
+    "The literal '*' means trust everything.",
+)
+@click.option(
+    "--root-path",
+    type=str,
+    default="",
+    help="Set the ASGI 'root_path' for applications submounted below a given URL path.",
+)
+@click.option(
+    "--limit-concurrency",
+    type=int,
+    default=None,
+    help="Maximum number of concurrent connections or tasks to allow, before issuing HTTP 503 responses.",
+)
+@click.option(
+    "--backlog",
+    type=int,
+    default=2048,
+    help="Maximum number of connections to hold in backlog",
+)
+@click.option(
+    "--limit-max-requests",
+    type=int,
+    default=None,
+    help="Maximum number of requests to service before terminating the process.",
+)
+@click.option(
+    "--limit-max-requests-jitter",
+    type=int,
+    default=0,
+    help="Maximum jitter to add to limit_max_requests."
+    " Staggers worker restarts to avoid all workers restarting simultaneously.",
+    show_default=True,
+)
+@click.option(
+    "--timeout-keep-alive",
+    type=int,
+    default=5,
+    help="Close Keep-Alive connections if no new data is received within this timeout (in seconds).",
+    show_default=True,
+)
+@click.option(
+    "--timeout-graceful-shutdown",
+    type=int,
+    default=None,
+    help="Maximum number of seconds to wait for graceful shutdown.",
+)
+@click.option(
+    "--timeout-worker-healthcheck",
+    type=int,
+    default=5,
+    help="Maximum number of seconds to wait for a worker to respond to a healthcheck.",
+    show_default=True,
+)
+@click.option("--ssl-keyfile", type=str, default=None, help="SSL key file", show_default=True)
+@click.option(
+    "--ssl-certfile",
+    type=str,
+    default=None,
+    help="SSL certificate file",
+    show_default=True,
+)
+@click.option(
+    "--ssl-keyfile-password",
+    type=str,
+    default=None,
+    help="SSL keyfile password",
+    show_default=True,
+)
+@click.option(
+    "--ssl-version",
+    type=int,
+    default=int(SSL_PROTOCOL_VERSION),
+    help="SSL version to use (see stdlib ssl module's)",
+    show_default=True,
+)
+@click.option(
+    "--ssl-cert-reqs",
+    type=int,
+    default=int(ssl.CERT_NONE),
+    help="Whether client certificate is required (see stdlib ssl module's)",
+    show_default=True,
+)
+@click.option(
+    "--ssl-ca-certs",
+    type=str,
+    default=None,
+    help="CA certificates file",
+    show_default=True,
+)
+@click.option(
+    "--ssl-ciphers",
+    type=str,
+    default=None,
+    help="Ciphers to use (see stdlib ssl module's). Defaults to OpenSSL's safe defaults.",
+    show_default=True,
+)
+@click.option(
+    "--header",
+    "headers",
+    multiple=True,
+    help="Specify custom default HTTP response headers as a Name:Value pair",
+)
+@click.option(
+    "--version",
+    is_flag=True,
+    callback=print_version,
+    expose_value=False,
+    is_eager=True,
+    help="Display the uvicorn version and exit.",
+)
+@click.option(
+    "--app-dir",
+    default="",
+    show_default=True,
+    help="Look for APP in the specified directory, by adding this to the PYTHONPATH."
+    " Defaults to the current working directory.",
+)
+@click.option(
+    "--h11-max-incomplete-event-size",
+    "h11_max_incomplete_event_size",
+    type=int,
+    default=None,
+    help="For h11, the maximum number of bytes to buffer of an incomplete event.",
+)
+@click.option(
+    "--reset-contextvars",
+    is_flag=True,
+    default=False,
+    help="Run each ASGI request in a fresh contextvars.Context. Hides context set in the lifespan.",
+    show_default=True,
+)
+@click.option(
+    "--factory",
+    is_flag=True,
+    default=False,
+    help="Treat APP as an application factory, i.e. a () -> <ASGI app> callable.",
+    show_default=True,
+)
+def main(
+    app: str,
+    host: str,
+    port: int,
+    uds: str,
+    fd: int,
+    loop: LoopFactoryType | str,
+    http: HTTPProtocolType | str,
+    ws: WSProtocolType | str,
+    ws_max_size: int,
+    ws_max_queue: int,
+    ws_ping_interval: float,
+    ws_ping_timeout: float,
+    ws_per_message_deflate: bool,
+    lifespan: LifespanType,
+    interface: InterfaceType,
+    reload: bool,
+    reload_dirs: list[str],
+    reload_includes: list[str],
+    reload_excludes: list[str],
+    reload_delay: float,
+    workers: int,
+    env_file: str,
+    log_config: str,
+    log_level: str,
+    access_log: bool,
+    proxy_headers: bool,
+    server_header: bool,
+    date_header: bool,
+    forwarded_allow_ips: str,
+    root_path: str,
+    limit_concurrency: int,
+    backlog: int,
+    limit_max_requests: int,
+    limit_max_requests_jitter: int,
+    timeout_keep_alive: int,
+    timeout_graceful_shutdown: int | None,
+    timeout_worker_healthcheck: int,
+    ssl_keyfile: str,
+    ssl_certfile: str,
+    ssl_keyfile_password: str,
+    ssl_version: int,
+    ssl_cert_reqs: int,
+    ssl_ca_certs: str,
+    ssl_ciphers: str | None,
+    headers: list[str],
+    use_colors: bool,
+    app_dir: str,
+    h11_max_incomplete_event_size: int | None,
+    reset_contextvars: bool,
+    factory: bool,
+) -> None:
+    run(
+        app,
+        host=host,
+        port=port,
+        uds=uds,
+        fd=fd,
+        loop=loop,
+        http=http,
+        ws=ws,
+        ws_max_size=ws_max_size,
+        ws_max_queue=ws_max_queue,
+        ws_ping_interval=ws_ping_interval,
+        ws_ping_timeout=ws_ping_timeout,
+        ws_per_message_deflate=ws_per_message_deflate,
+        lifespan=lifespan,
+        env_file=env_file,
+        log_config=LOGGING_CONFIG if log_config is None else log_config,
+        log_level=log_level,
+        access_log=access_log,
+        interface=interface,
+        reload=reload,
+        reload_dirs=reload_dirs or None,
+        reload_includes=reload_includes or None,
+        reload_excludes=reload_excludes or None,
+        reload_delay=reload_delay,
+        workers=workers,
+        proxy_headers=proxy_headers,
+        server_header=server_header,
+        date_header=date_header,
+        forwarded_allow_ips=forwarded_allow_ips,
+        root_path=root_path,
+        limit_concurrency=limit_concurrency,
+        backlog=backlog,
+        limit_max_requests=limit_max_requests,
+        limit_max_requests_jitter=limit_max_requests_jitter,
+        timeout_keep_alive=timeout_keep_alive,
+        timeout_graceful_shutdown=timeout_graceful_shutdown,
+        timeout_worker_healthcheck=timeout_worker_healthcheck,
+        ssl_keyfile=ssl_keyfile,
+        ssl_certfile=ssl_certfile,
+        ssl_keyfile_password=ssl_keyfile_password,
+        ssl_version=ssl_version,
+        ssl_cert_reqs=ssl_cert_reqs,
+        ssl_ca_certs=ssl_ca_certs,
+        ssl_ciphers=ssl_ciphers,
+        headers=[header.split(":", 1) for header in headers],  # type: ignore[misc]
+        use_colors=use_colors,
+        factory=factory,
+        app_dir=app_dir,
+        h11_max_incomplete_event_size=h11_max_incomplete_event_size,
+        reset_contextvars=reset_contextvars,
+    )
+
+
+def run(
+    app: ASGIApplication | Callable[..., Any] | str,
+    *,
+    host: str = "127.0.0.1",
+    port: int = 8000,
+    uds: str | None = None,
+    fd: int | None = None,
+    loop: LoopFactoryType | str = "auto",
+    http: type[asyncio.Protocol] | HTTPProtocolType | str = "auto",
+    ws: type[asyncio.Protocol] | WSProtocolType | str = "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: LifespanType = "auto",
+    interface: InterfaceType = "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[str] | None = None,
+    log_config: dict[str, Any] | str | os.PathLike[str] | RawConfigParser | 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,
+    limit_max_requests_jitter: int = 0,
+    timeout_keep_alive: int = 5,
+    timeout_graceful_shutdown: int | None = None,
+    timeout_worker_healthcheck: int = 5,
+    ssl_keyfile: str | os.PathLike[str] | None = None,
+    ssl_certfile: str | os.PathLike[str] | None = None,
+    ssl_keyfile_password: str | None = None,
+    ssl_version: int = SSL_PROTOCOL_VERSION,
+    ssl_cert_reqs: int = ssl.CERT_NONE,
+    ssl_ca_certs: str | os.PathLike[str] | None = None,
+    ssl_ciphers: str | None = None,
+    ssl_context_factory: Callable[[Config, Callable[[], ssl.SSLContext]], ssl.SSLContext] | None = None,
+    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,
+    reset_contextvars: bool = False,
+) -> None:
+    if app_dir is not None:
+        sys.path.insert(0, app_dir)
+
+    config = Config(
+        app,
+        host=host,
+        port=port,
+        uds=uds,
+        fd=fd,
+        loop=loop,
+        http=http,
+        ws=ws,
+        ws_max_size=ws_max_size,
+        ws_max_queue=ws_max_queue,
+        ws_ping_interval=ws_ping_interval,
+        ws_ping_timeout=ws_ping_timeout,
+        ws_per_message_deflate=ws_per_message_deflate,
+        lifespan=lifespan,
+        interface=interface,
+        reload=reload,
+        reload_dirs=reload_dirs,
+        reload_includes=reload_includes,
+        reload_excludes=reload_excludes,
+        reload_delay=reload_delay,
+        workers=workers,
+        env_file=env_file,
+        log_config=log_config,
+        log_level=log_level,
+        access_log=access_log,
+        proxy_headers=proxy_headers,
+        server_header=server_header,
+        date_header=date_header,
+        forwarded_allow_ips=forwarded_allow_ips,
+        root_path=root_path,
+        limit_concurrency=limit_concurrency,
+        backlog=backlog,
+        limit_max_requests=limit_max_requests,
+        limit_max_requests_jitter=limit_max_requests_jitter,
+        timeout_keep_alive=timeout_keep_alive,
+        timeout_graceful_shutdown=timeout_graceful_shutdown,
+        timeout_worker_healthcheck=timeout_worker_healthcheck,
+        ssl_keyfile=ssl_keyfile,
+        ssl_certfile=ssl_certfile,
+        ssl_keyfile_password=ssl_keyfile_password,
+        ssl_version=ssl_version,
+        ssl_cert_reqs=ssl_cert_reqs,
+        ssl_ca_certs=ssl_ca_certs,
+        ssl_ciphers=ssl_ciphers,
+        ssl_context_factory=ssl_context_factory,
+        headers=headers,
+        use_colors=use_colors,
+        factory=factory,
+        h11_max_incomplete_event_size=h11_max_incomplete_event_size,
+        reset_contextvars=reset_contextvars,
+    )
+    if (config.reload or config.workers > 1) and not isinstance(app, str):
+        logger = logging.getLogger("uvicorn.error")
+        logger.warning("You must pass the application as an import string to enable 'reload' or 'workers'.")
+        sys.exit(1)
+
+    config.load_app()
+    server = Server(config=config)
+
+    try:
+        if config.should_reload:
+            sock = config.bind_socket()
+            ChangeReload(config, target=server.run, sockets=[sock]).run()
+        elif config.workers > 1:
+            sock = config.bind_socket()
+            Multiprocess(config, target=server.run, sockets=[sock]).run()
+        else:
+            server.run()
+    except KeyboardInterrupt:  # pragma: full coverage
+        pass
+    finally:
+        if config.uds and os.path.exists(config.uds):
+            os.remove(config.uds)  # pragma: py-win32
+
+    if not server.started and not config.should_reload and config.workers == 1:
+        sys.exit(STARTUP_FAILURE)
+
+
+def __getattr__(name: str) -> Any:
+    if name == "ServerState":
+        warnings.warn(
+            "uvicorn.main.ServerState is deprecated, use uvicorn.server.ServerState instead.",
+            DeprecationWarning,
+        )
+        from uvicorn.server import ServerState
+
+        return ServerState
+    raise AttributeError(f"module {__name__} has no attribute {name}")
+
+
+if __name__ == "__main__":
+    main()  # pragma: no cover