Remove asgiref dependency

requirements.txt

@@ -1,5 +1,8 @@
 -e .[standard]
 
+# Type annotation
+asgiref==3.5.2
+
 # Explicit optionals
 wsproto==1.1.0
 

setup.py

@@ -44,7 +44,6 @@ def get_packages(package):
 env_marker_below_38 = "python_version < '3.8'"
 
 minimal_requirements = [
-    "asgiref>=3.4.0",
     "click>=7.0",
     "h11>=0.8",
     "typing-extensions;" + env_marker_below_38,

tests/middleware/test_wsgi.py

@@ -1,14 +1,16 @@
 import io
 import sys
-from typing import AsyncGenerator, List
+from typing import TYPE_CHECKING, AsyncGenerator, List
 
 import httpx
 import pytest
-from asgiref.typing import HTTPRequestEvent, HTTPScope
 
 from uvicorn._types import Environ, StartResponse
 from uvicorn.middleware.wsgi import WSGIMiddleware, build_environ
 
+if TYPE_CHECKING:
+    from asgiref.typing import HTTPRequestEvent, HTTPScope
+
 
 def hello_world(environ: Environ, start_response: StartResponse) -> List[bytes]:
     status = "200 OK"
@@ -114,7 +116,7 @@ async def test_wsgi_exc_info() -> None:
 
 
 def test_build_environ_encoding() -> None:
-    scope: HTTPScope = {
+    scope: "HTTPScope" = {
         "asgi": {"version": "3.0", "spec_version": "2.0"},
         "scheme": "http",
         "raw_path": b"/\xe6\x96\x87",
@@ -129,7 +131,7 @@ def test_build_environ_encoding() -> None:
         "headers": [(b"key", b"value1"), (b"key", b"value2")],
         "extensions": {},
     }
-    message: HTTPRequestEvent = {
+    message: "HTTPRequestEvent" = {
         "type": "http.request",
         "body": b"",
         "more_body": False,

tests/supervisors/test_multiprocess.py

@@ -1,14 +1,17 @@
 import signal
 import socket
-from typing import List, Optional
-
-from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, Scope
+from typing import TYPE_CHECKING, List, Optional
 
 from uvicorn import Config
 from uvicorn.supervisors import Multiprocess
 
+if TYPE_CHECKING:
+    from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, Scope
+
 
-def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
+def app(
+    scope: "Scope", receive: "ASGIReceiveCallable", send: "ASGISendCallable"
+) -> None:
     pass  # pragma: no cover
 
 

tests/test_config.py

@@ -7,14 +7,8 @@
 from pathlib import Path
 from unittest.mock import MagicMock
 
-if sys.version_info < (3, 8):  # pragma: py-gte-38
-    from typing_extensions import Literal
-else:  # pragma: py-lt-38
-    from typing import Literal
-
 import pytest
 import yaml
-from asgiref.typing import ASGIApplication, ASGIReceiveCallable, ASGISendCallable, Scope
 from pytest_mock import MockerFixture
 
 from tests.utils import as_cwd
@@ -25,6 +19,19 @@
 from uvicorn.middleware.wsgi import WSGIMiddleware
 from uvicorn.protocols.http.h11_impl import H11Protocol
 
+if sys.version_info < (3, 8):  # pragma: py-gte-38
+    from typing_extensions import Literal
+else:  # pragma: py-lt-38
+    from typing import Literal
+
+if typing.TYPE_CHECKING:
+    from asgiref.typing import (
+        ASGIApplication,
+        ASGIReceiveCallable,
+        ASGISendCallable,
+        Scope,
+    )
+
 
 @pytest.fixture
 def mocked_logging_config_module(mocker: MockerFixture) -> MagicMock:
@@ -42,7 +49,7 @@ def yaml_logging_config(logging_config: dict) -> str:
 
 
 async def asgi_app(
-    scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    scope: "Scope", receive: "ASGIReceiveCallable", send: "ASGISendCallable"
 ) -> None:
     pass  # pragma: nocover
 
@@ -64,7 +71,7 @@ def test_debug_app() -> None:
     [(asgi_app, False), ("tests.test_config:asgi_app", True)],
 )
 def test_config_should_reload_is_set(
-    app: ASGIApplication, expected_should_reload: bool
+    app: "ASGIApplication", expected_should_reload: bool
 ) -> None:
     config_debug = Config(app=app, debug=True)
     assert config_debug.debug is True
@@ -269,7 +276,7 @@ def test_app_unimportable_other(caplog: pytest.LogCaptureFixture) -> None:
 
 
 def test_app_factory(caplog: pytest.LogCaptureFixture) -> None:
-    def create_app() -> ASGIApplication:
+    def create_app() -> "ASGIApplication":
         return asgi_app
 
     config = Config(app=create_app, factory=True, proxy_headers=False)
@@ -330,9 +337,9 @@ def test_ssl_config_combined(tls_certificate_key_and_chain_path: str) -> None:
     assert config.is_ssl is True
 
 
-def asgi2_app(scope: Scope) -> typing.Callable:
+def asgi2_app(scope: "Scope") -> typing.Callable:
     async def asgi(
-        receive: ASGIReceiveCallable, send: ASGISendCallable
+        receive: "ASGIReceiveCallable", send: "ASGISendCallable"
     ) -> None:  # pragma: nocover
         pass
 
@@ -343,7 +350,7 @@ async def asgi(
     "app, expected_interface", [(asgi_app, "3.0"), (asgi2_app, "2.0")]
 )
 def test_asgi_version(
-    app: ASGIApplication, expected_interface: Literal["2.0", "3.0"]
+    app: "ASGIApplication", expected_interface: Literal["2.0", "3.0"]
 ) -> None:
     config = Config(app=app)
     config.load()

tests/test_subprocess.py

@@ -1,21 +1,23 @@
 import socket
 import sys
-from typing import List
+from typing import TYPE_CHECKING, List
 from unittest.mock import patch
 
 import pytest
-from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, Scope
 
 from uvicorn._subprocess import SpawnProcess, get_subprocess, subprocess_started
 from uvicorn.config import Config
 
+if TYPE_CHECKING:
+    from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, Scope
+
 
 def server_run(sockets: List[socket.socket]):  # pragma: no cover
     ...
 
 
 async def app(
-    scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    scope: "Scope", receive: "ASGIReceiveCallable", send: "ASGISendCallable"
 ) -> None:  # pragma: no cover
     ...
 

uvicorn/config.py

@@ -8,7 +8,17 @@
 import ssl
 import sys
 from pathlib import Path
-from typing import Awaitable, Callable, Dict, List, Optional, Tuple, Type, Union
+from typing import (
+    TYPE_CHECKING,
+    Awaitable,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    Type,
+    Union,
+)
 
 from h11._connection import DEFAULT_MAX_INCOMPLETE_EVENT_SIZE
 
@@ -20,7 +30,6 @@
     from typing import Literal
 
 import click
-from asgiref.typing import ASGIApplication
 
 try:
     import yaml
@@ -37,6 +46,9 @@
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 from uvicorn.middleware.wsgi import WSGIMiddleware
 
+if TYPE_CHECKING:
+    from asgiref.typing import ASGIApplication
+
 HTTPProtocolType = Literal["auto", "h11", "httptools"]
 WSProtocolType = Literal["auto", "none", "websockets", "wsproto"]
 LifespanType = Literal["auto", "on", "off"]
@@ -196,7 +208,7 @@ def _normalize_dirs(dirs: Union[List[str], str, None]) -> List[str]:
 class Config:
     def __init__(
         self,
-        app: Union[ASGIApplication, Callable, str],
+        app: Union["ASGIApplication", Callable, str],
         host: str = "127.0.0.1",
         port: int = 8000,
         uds: Optional[str] = None,

uvicorn/lifespan/on.py

@@ -1,27 +1,29 @@
 import asyncio
 import logging
 from asyncio import Queue
-from typing import Union
-
-from asgiref.typing import (
-    LifespanScope,
-    LifespanShutdownCompleteEvent,
-    LifespanShutdownEvent,
-    LifespanShutdownFailedEvent,
-    LifespanStartupCompleteEvent,
-    LifespanStartupEvent,
-    LifespanStartupFailedEvent,
-)
+from typing import TYPE_CHECKING, Union
 
 from uvicorn import Config
 
-LifespanReceiveMessage = Union[LifespanStartupEvent, LifespanShutdownEvent]
-LifespanSendMessage = Union[
-    LifespanStartupFailedEvent,
-    LifespanShutdownFailedEvent,
-    LifespanStartupCompleteEvent,
-    LifespanShutdownCompleteEvent,
-]
+if TYPE_CHECKING:
+    from asgiref.typing import (
+        LifespanScope,
+        LifespanShutdownCompleteEvent,
+        LifespanShutdownEvent,
+        LifespanShutdownFailedEvent,
+        LifespanStartupCompleteEvent,
+        LifespanStartupEvent,
+        LifespanStartupFailedEvent,
+    )
+
+    LifespanReceiveMessage = Union[LifespanStartupEvent, LifespanShutdownEvent]
+    LifespanSendMessage = Union[
+        LifespanStartupFailedEvent,
+        LifespanShutdownFailedEvent,
+        LifespanStartupCompleteEvent,
+        LifespanShutdownCompleteEvent,
+    ]
+
 
 STATE_TRANSITION_ERROR = "Got invalid state transition on lifespan protocol."
 
@@ -97,7 +99,7 @@ async def main(self) -> None:
             self.startup_event.set()
             self.shutdown_event.set()
 
-    async def send(self, message: LifespanSendMessage) -> None:
+    async def send(self, message: "LifespanSendMessage") -> None:
         assert message["type"] in (
             "lifespan.startup.complete",
             "lifespan.startup.failed",
@@ -131,5 +133,5 @@ async def send(self, message: LifespanSendMessage) -> None:
             if message.get("message"):
                 self.logger.error(message["message"])
 
-    async def receive(self) -> LifespanReceiveMessage:
+    async def receive(self) -> "LifespanReceiveMessage":
         return await self.receive_queue.get()

uvicorn/main.py

@@ -6,7 +6,6 @@
 import typing
 
 import click
-from asgiref.typing import ASGIApplication
 from h11._connection import DEFAULT_MAX_INCOMPLETE_EVENT_SIZE
 
 import uvicorn
@@ -29,6 +28,9 @@
 from uvicorn.server import Server, ServerState  # noqa: F401  # Used to be defined here.
 from uvicorn.supervisors import ChangeReload, Multiprocess
 
+if typing.TYPE_CHECKING:
+    from asgiref.typing import ASGIApplication
+
 LEVEL_CHOICES = click.Choice(list(LOG_LEVELS.keys()))
 HTTP_CHOICES = click.Choice(list(HTTP_PROTOCOLS.keys()))
 WS_CHOICES = click.Choice(list(WS_PROTOCOLS.keys()))
@@ -453,7 +455,7 @@ def main(
 
 
 def run(
-    app: typing.Union[ASGIApplication, str],
+    app: typing.Union["ASGIApplication", str],
     *,
     host: str = "127.0.0.1",
     port: int = 8000,

uvicorn/middleware/asgi2.py

@@ -1,17 +1,20 @@
-from asgiref.typing import (
-    ASGI2Application,
-    ASGIReceiveCallable,
-    ASGISendCallable,
-    Scope,
-)
+import typing
+
+if typing.TYPE_CHECKING:
+    from asgiref.typing import (
+        ASGI2Application,
+        ASGIReceiveCallable,
+        ASGISendCallable,
+        Scope,
+    )
 
 
 class ASGI2Middleware:
-    def __init__(self, app: ASGI2Application):
+    def __init__(self, app: "ASGI2Application"):
         self.app = app
 
     async def __call__(
-        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
+        self, scope: "Scope", receive: "ASGIReceiveCallable", send: "ASGISendCallable"
     ) -> None:
         instance = self.app(scope)
         await instance(receive, send)