codekingpro/portable-devtools

1import sys2from typing import (3    Any,4    Awaitable,5    Callable,6    Dict,7    Iterable,8    Literal,9    Optional,10    Protocol,11    Tuple,12    Type,13    TypedDict,14    Union,15)16 17if sys.version_info >= (3, 11):18    from typing import NotRequired19else:20    from typing_extensions import NotRequired21 22__all__ = (23    "ASGIVersions",24    "HTTPScope",25    "WebSocketScope",26    "LifespanScope",27    "WWWScope",28    "Scope",29    "HTTPRequestEvent",30    "HTTPResponseStartEvent",31    "HTTPResponseBodyEvent",32    "HTTPResponseTrailersEvent",33    "HTTPResponsePathsendEvent",34    "HTTPServerPushEvent",35    "HTTPDisconnectEvent",36    "WebSocketConnectEvent",37    "WebSocketAcceptEvent",38    "WebSocketReceiveEvent",39    "WebSocketSendEvent",40    "WebSocketResponseStartEvent",41    "WebSocketResponseBodyEvent",42    "WebSocketDisconnectEvent",43    "WebSocketCloseEvent",44    "LifespanStartupEvent",45    "LifespanShutdownEvent",46    "LifespanStartupCompleteEvent",47    "LifespanStartupFailedEvent",48    "LifespanShutdownCompleteEvent",49    "LifespanShutdownFailedEvent",50    "ASGIReceiveEvent",51    "ASGISendEvent",52    "ASGIReceiveCallable",53    "ASGISendCallable",54    "ASGI2Protocol",55    "ASGI2Application",56    "ASGI3Application",57    "ASGIApplication",58)59 60 61class ASGIVersions(TypedDict):62    spec_version: str63    version: Union[Literal["2.0"], Literal["3.0"]]64 65 66class HTTPScope(TypedDict):67    type: Literal["http"]68    asgi: ASGIVersions69    http_version: str70    method: str71    scheme: str72    path: str73    raw_path: bytes74    query_string: bytes75    root_path: str76    headers: Iterable[Tuple[bytes, bytes]]77    client: Optional[Tuple[str, int]]78    server: Optional[Tuple[str, Optional[int]]]79    state: NotRequired[Dict[str, Any]]80    extensions: Optional[Dict[str, Dict[object, object]]]81 82 83class WebSocketScope(TypedDict):84    type: Literal["websocket"]85    asgi: ASGIVersions86    http_version: str87    scheme: str88    path: str89    raw_path: bytes90    query_string: bytes91    root_path: str92    headers: Iterable[Tuple[bytes, bytes]]93    client: Optional[Tuple[str, int]]94    server: Optional[Tuple[str, Optional[int]]]95    subprotocols: Iterable[str]96    state: NotRequired[Dict[str, Any]]97    extensions: Optional[Dict[str, Dict[object, object]]]98 99 100class LifespanScope(TypedDict):101    type: Literal["lifespan"]102    asgi: ASGIVersions103    state: NotRequired[Dict[str, Any]]104 105 106WWWScope = Union[HTTPScope, WebSocketScope]107Scope = Union[HTTPScope, WebSocketScope, LifespanScope]108 109 110class HTTPRequestEvent(TypedDict):111    type: Literal["http.request"]112    body: bytes113    more_body: bool114 115 116class HTTPResponseDebugEvent(TypedDict):117    type: Literal["http.response.debug"]118    info: Dict[str, object]119 120 121class HTTPResponseStartEvent(TypedDict):122    type: Literal["http.response.start"]123    status: int124    headers: Iterable[Tuple[bytes, bytes]]125    trailers: bool126 127 128class HTTPResponseBodyEvent(TypedDict):129    type: Literal["http.response.body"]130    body: bytes131    more_body: bool132 133 134class HTTPResponseTrailersEvent(TypedDict):135    type: Literal["http.response.trailers"]136    headers: Iterable[Tuple[bytes, bytes]]137    more_trailers: bool138 139 140class HTTPResponsePathsendEvent(TypedDict):141    type: Literal["http.response.pathsend"]142    path: str143 144 145class HTTPServerPushEvent(TypedDict):146    type: Literal["http.response.push"]147    path: str148    headers: Iterable[Tuple[bytes, bytes]]149 150 151class HTTPDisconnectEvent(TypedDict):152    type: Literal["http.disconnect"]153 154 155class WebSocketConnectEvent(TypedDict):156    type: Literal["websocket.connect"]157 158 159class WebSocketAcceptEvent(TypedDict):160    type: Literal["websocket.accept"]161    subprotocol: Optional[str]162    headers: Iterable[Tuple[bytes, bytes]]163 164 165class WebSocketReceiveEvent(TypedDict):166    type: Literal["websocket.receive"]167    bytes: Optional[bytes]168    text: Optional[str]169 170 171class WebSocketSendEvent(TypedDict):172    type: Literal["websocket.send"]173    bytes: Optional[bytes]174    text: Optional[str]175 176 177class WebSocketResponseStartEvent(TypedDict):178    type: Literal["websocket.http.response.start"]179    status: int180    headers: Iterable[Tuple[bytes, bytes]]181 182 183class WebSocketResponseBodyEvent(TypedDict):184    type: Literal["websocket.http.response.body"]185    body: bytes186    more_body: bool187 188 189class WebSocketDisconnectEvent(TypedDict):190    type: Literal["websocket.disconnect"]191    code: int192    reason: Optional[str]193 194 195class WebSocketCloseEvent(TypedDict):196    type: Literal["websocket.close"]197    code: int198    reason: Optional[str]199 200 201class LifespanStartupEvent(TypedDict):202    type: Literal["lifespan.startup"]203 204 205class LifespanShutdownEvent(TypedDict):206    type: Literal["lifespan.shutdown"]207 208 209class LifespanStartupCompleteEvent(TypedDict):210    type: Literal["lifespan.startup.complete"]211 212 213class LifespanStartupFailedEvent(TypedDict):214    type: Literal["lifespan.startup.failed"]215    message: str216 217 218class LifespanShutdownCompleteEvent(TypedDict):219    type: Literal["lifespan.shutdown.complete"]220 221 222class LifespanShutdownFailedEvent(TypedDict):223    type: Literal["lifespan.shutdown.failed"]224    message: str225 226 227ASGIReceiveEvent = Union[228    HTTPRequestEvent,229    HTTPDisconnectEvent,230    WebSocketConnectEvent,231    WebSocketReceiveEvent,232    WebSocketDisconnectEvent,233    LifespanStartupEvent,234    LifespanShutdownEvent,235]236 237 238ASGISendEvent = Union[239    HTTPResponseStartEvent,240    HTTPResponseBodyEvent,241    HTTPResponseTrailersEvent,242    HTTPServerPushEvent,243    HTTPDisconnectEvent,244    WebSocketAcceptEvent,245    WebSocketSendEvent,246    WebSocketResponseStartEvent,247    WebSocketResponseBodyEvent,248    WebSocketCloseEvent,249    LifespanStartupCompleteEvent,250    LifespanStartupFailedEvent,251    LifespanShutdownCompleteEvent,252    LifespanShutdownFailedEvent,253]254 255 256ASGIReceiveCallable = Callable[[], Awaitable[ASGIReceiveEvent]]257ASGISendCallable = Callable[[ASGISendEvent], Awaitable[None]]258 259 260class ASGI2Protocol(Protocol):261    def __init__(self, scope: Scope) -> None:262        ...263 264    async def __call__(265        self, receive: ASGIReceiveCallable, send: ASGISendCallable266    ) -> None:267        ...268 269 270ASGI2Application = Type[ASGI2Protocol]271ASGI3Application = Callable[272    [273        Scope,274        ASGIReceiveCallable,275        ASGISendCallable,276    ],277    Awaitable[None],278]279ASGIApplication = Union[ASGI2Application, ASGI3Application]280