create context var for current atomic operation

Author: mahenzon

fastapi_jsonapi/atomic/__init__.py

@@ -1,3 +1,7 @@
-__all__ = ("AtomicOperations",)
+__all__ = (
+    "AtomicOperations",
+    "current_atomic_operation",
+)
 
 from .atomic import AtomicOperations
+from .atomic_handler import current_atomic_operation

fastapi_jsonapi/atomic/atomic_handler.py

@@ -2,6 +2,7 @@
 
 import logging
 from collections import defaultdict
+from contextvars import ContextVar
 from functools import wraps
 from typing import (
     TYPE_CHECKING,
@@ -28,6 +29,8 @@
 log = logging.getLogger(__name__)
 AtomicResponseDict = TypedDict("AtomicResponseDict", {"atomic:results": List[Any]})
 
+current_atomic_operation: ContextVar[OperationBase] = ContextVar("current_atomic_operation")
+
 
 def catch_exc_on_operation_handle(func: Callable[..., Awaitable]):
     @wraps(func)
@@ -82,7 +85,7 @@ async def prepare_one_operation(self, operation: AtomicOperation):
             raise ValueError(msg)
         jsonapi = RoutersJSONAPI.all_jsonapi_routers[operation_type]
 
-        one_operation = await OperationBase.prepare(
+        one_operation = OperationBase.prepare(
             action=operation.op,
             request=self.request,
             jsonapi=jsonapi,
@@ -116,6 +119,9 @@ async def handle(self) -> Union[AtomicResponseDict, AtomicResultResponse, None]:
         success = True
         previous_dl: Optional[BaseDataLayer] = None
         for operation in prepared_operations:
+            # set context var
+            ctx_var_token = current_atomic_operation.set(operation)
+
             dl: BaseDataLayer = await operation.get_data_layer()
             await dl.atomic_start(previous_dl=previous_dl)
             response = await self.process_one_operation(
@@ -136,6 +142,9 @@ async def handle(self) -> Union[AtomicResponseDict, AtomicResultResponse, None]:
             if operation.data.lid and response.data:
                 self.local_ids_cache[operation.data.type][operation.data.lid] = response.data.id
 
+            # reset context var
+            current_atomic_operation.reset(ctx_var_token)
+
         if previous_dl:
             await previous_dl.atomic_end(success=success)
 

fastapi_jsonapi/atomic/prepared_atomic_operation.py

@@ -24,11 +24,14 @@ class OperationBase:
     view: ViewBase
     ref: Optional[AtomicOperationRef]
     data: OperationDataType
-    data_layer_view_dependencies: Dict[str, Any]
     op_type: str
 
+    @property
+    def http_method(self) -> HTTPMethod:
+        raise NotImplementedError
+
     @classmethod
-    async def prepare(
+    def prepare(
         cls,
         action: str,
         request: Request,
@@ -55,22 +58,21 @@ async def prepare(
 
         view = view_cls(request=request, jsonapi=jsonapi)
 
-        data_layer_view_dependencies: Dict[str, Any] = await jsonapi.handle_view_dependencies(
-            request=request,
-            view_cls=view_cls,
-            method=operation_cls.http_method,
-        )
         return operation_cls(
             jsonapi=jsonapi,
             view=view,
             ref=ref,
             data=data,
-            data_layer_view_dependencies=data_layer_view_dependencies,
             op_type=action,
         )
 
     async def get_data_layer(self) -> BaseDataLayer:
-        return await self.view.get_data_layer(self.data_layer_view_dependencies)
+        data_layer_view_dependencies: Dict[str, Any] = await self.jsonapi.handle_view_dependencies(
+            request=self.view.request,
+            view_cls=self.view.__class__,
+            method=self.http_method,
+        )
+        return await self.view.get_data_layer(data_layer_view_dependencies)
 
     async def handle(self, dl: BaseDataLayer):
         raise NotImplementedError
@@ -135,6 +137,9 @@ class OperationAdd(ListOperationBase):
     http_method = HTTPMethod.POST
 
     async def handle(self, dl: BaseDataLayer):
+        # use outer schema wrapper because we need this error path:
+        # `{'loc': ['data', 'attributes', 'name']`
+        # and not `{'loc': ['attributes', 'name']`
         data_in = self.jsonapi.schema_in_post(data=self.data)
         response = await self.view.process_create_object(
             dl=dl,
@@ -151,6 +156,10 @@ async def handle(self, dl: BaseDataLayer):
             # TODO: clear to-one relationships
             pass
         # TODO: handle relationship update requests (relationship resources)
+
+        # use outer schema wrapper because we need this error path:
+        # `{'loc': ['data', 'attributes', 'name']`
+        # and not `{'loc': ['attributes', 'name']`
         data_in = self.jsonapi.schema_in_patch(data=self.data)
         obj_id = self.ref and self.ref.id or self.data and self.data.id
         response = await self.view.process_update_object(