diff --git a/setup.py b/setup.py index 87d20fac54..cb1c1cfdde 100755 --- a/setup.py +++ b/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, diff --git a/uvicorn/_types.py b/uvicorn/_types.py index 0a547a50d3..96d884c3d3 100644 --- a/uvicorn/_types.py +++ b/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] diff --git a/uvicorn/config.py b/uvicorn/config.py index 681ca0caf9..26605cf912 100644 --- a/uvicorn/config.py +++ b/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: diff --git a/uvicorn/lifespan/on.py b/uvicorn/lifespan/on.py index caf6e505ec..ec68189cd0 100644 --- a/uvicorn/lifespan/on.py +++ b/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 ( diff --git a/uvicorn/protocols/utils.py b/uvicorn/protocols/utils.py index 1da733600a..637122bc1a 100644 --- a/uvicorn/protocols/utils.py +++ b/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]]: