Revert "♻️ Replace _types module by asgiref.typing (encode#1044)" (re…

…move asgiref constraint)

This reverts commit 926a8f5.

setup.py

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

uvicorn/_types.py

@@ -1,5 +1,13 @@
 import types
 import typing
+import sys
+from typing import Dict, Iterable, Optional, Tuple, Union
+
+if sys.version_info < (3, 8):
+    from typing_extensions import Literal, TypedDict
+else:
+    from typing import Literal, TypedDict
+
 
 # WSGI
 Environ = typing.MutableMapping[str, typing.Any]
@@ -12,3 +20,62 @@
 WSGIApp = typing.Callable[
     [Environ, StartResponse], typing.Union[typing.Iterable[bytes], BaseException]
 ]
+
+
+class ASGISpecInfo(TypedDict):
+    version: str
+    spec_version: Optional[Literal["2.0", "2.1"]]
+
+
+class LifespanScope(TypedDict):
+    type: Literal["lifespan"]
+    asgi: ASGISpecInfo
+
+
+class LifespanReceiveMessage(TypedDict):
+    type: Literal["lifespan.startup", "lifespan.shutdown"]
+
+
+class LifespanSendMessage(TypedDict):
+    type: Literal[
+        "lifespan.startup.complete",
+        "lifespan.startup.failed",
+        "lifespan.shutdown.complete",
+        "lifespan.shutdown.failed",
+    ]
+    message: Optional[str]
+
+
+class HTTPScope(TypedDict):
+    type: Literal["http"]
+    asgi: ASGISpecInfo
+    http_version: str
+    method: str
+    scheme: str
+    path: str
+    raw_path: bytes
+    query_string: bytes
+    root_path: str
+    headers: Iterable[Tuple[bytes, bytes]]
+    client: Optional[Tuple[str, int]]
+    server: Optional[Tuple[str, Optional[int]]]
+    extensions: Dict[str, Dict[object, object]]
+
+
+class WebsocketScope(TypedDict):
+    type: Literal["websocket"]
+    asgi: ASGISpecInfo
+    http_version: str
+    scheme: str
+    path: str
+    raw_path: bytes
+    query_string: bytes
+    root_path: str
+    headers: Iterable[Tuple[bytes, bytes]]
+    client: Optional[Tuple[str, int]]
+    server: Optional[Tuple[str, Optional[int]]]
+    subprotocols: Iterable[str]
+    extensions: Dict[str, Dict[object, object]]
+
+
+WWWScope = Union[HTTPScope, WebsocketScope]

uvicorn/config.py

@@ -361,13 +361,8 @@ def __init__(
             self.forwarded_allow_ips = forwarded_allow_ips
 
     @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]
+    def asgi_version(self) -> str:
+        return {"asgi2": "2.0", "asgi3": "3.0", "wsgi": "3.0"}[self.interface]
 
     @property
     def is_ssl(self) -> bool:

uvicorn/lifespan/on.py

@@ -1,27 +1,9 @@
 import asyncio
 import logging
 from asyncio import Queue
-from typing import Union
-
-from asgiref.typing import (
-    LifespanScope,
-    LifespanShutdownCompleteEvent,
-    LifespanShutdownEvent,
-    LifespanShutdownFailedEvent,
-    LifespanStartupCompleteEvent,
-    LifespanStartupEvent,
-    LifespanStartupFailedEvent,
-)
 
 from uvicorn import Config
-
-LifespanReceiveMessage = Union[LifespanStartupEvent, LifespanShutdownEvent]
-LifespanSendMessage = Union[
-    LifespanStartupFailedEvent,
-    LifespanShutdownFailedEvent,
-    LifespanStartupCompleteEvent,
-    LifespanShutdownCompleteEvent,
-]
+from uvicorn._types import LifespanReceiveMessage, LifespanScope, LifespanSendMessage
 
 STATE_TRANSITION_ERROR = "Got invalid state transition on lifespan protocol."
 
@@ -48,8 +30,8 @@ async def startup(self) -> None:
         main_lifespan_task = loop.create_task(self.main())  # noqa: F841
         # Keep a hard reference to prevent garbage collection
         # See https://github.com/encode/uvicorn/pull/972
-        startup_event: LifespanStartupEvent = {"type": "lifespan.startup"}
-        await self.receive_queue.put(startup_event)
+
+        await self.receive_queue.put({"type": "lifespan.startup"})
         await self.startup_event.wait()
 
         if self.startup_failed or (self.error_occured and self.config.lifespan == "on"):
@@ -62,8 +44,7 @@ async def shutdown(self) -> None:
         if self.error_occured:
             return
         self.logger.info("Waiting for application shutdown.")
-        shutdown_event: LifespanShutdownEvent = {"type": "lifespan.shutdown"}
-        await self.receive_queue.put(shutdown_event)
+        await self.receive_queue.put({"type": "lifespan.shutdown"})
         await self.shutdown_event.wait()
 
         if self.shutdown_failed or (

uvicorn/protocols/utils.py

@@ -2,7 +2,7 @@
 import urllib.parse
 from typing import Optional, Tuple
 
-from asgiref.typing import WWWScope
+from uvicorn._types import WWWScope
 
 
 def get_remote_addr(transport: asyncio.Transport) -> Optional[Tuple[str, int]]: