Moved from a thread local to contextvar context for tracking asgi route handler args

Author: eadwinCode

ellar/asgi_args.py

@@ -0,0 +1,19 @@
+import typing as t
+
+from ellar.types import TReceive, TScope, TSend
+
+if t.TYPE_CHECKING:
+    from ellar.di.providers import Provider
+
+
+class ASGIArgs:
+    __slots__ = ("scope", "receive", "send", "context")
+
+    def __init__(self, scope: TScope, receive: TReceive, send: TSend) -> None:
+        self.scope = scope
+        self.receive = receive
+        self.send = send
+        self.context: t.Dict[t.Type, "Provider"] = {}
+
+    def get_args(self) -> t.Tuple[TScope, TReceive, TSend]:
+        return self.scope, self.receive, self.send

ellar/constants.py

@@ -1,6 +1,7 @@
+import contextvars
 import logging
 from enum import Enum
-from typing import Any, Dict, List, no_type_check
+from typing import Any, Dict, List, Optional, no_type_check
 
 from pydantic.fields import (
     SHAPE_LIST,
@@ -10,6 +11,13 @@
     SHAPE_TUPLE_ELLIPSIS,
 )
 
+from .asgi_args import ASGIArgs
+
+ASGI_CONTEXT_VAR: contextvars.ContextVar[Optional[ASGIArgs]] = contextvars.ContextVar(
+    "ASGI-CONTEXT-VAR"
+)
+ASGI_CONTEXT_VAR.set(None)
+
 
 class _AnnotationToValue(type):
     keys: List[str]