create atomic operations base

Author: mahenzon

examples/api_for_sqlalchemy/urls.py

@@ -20,6 +20,7 @@
     UserBio,
 )
 from fastapi_jsonapi import RoutersJSONAPI
+from fastapi_jsonapi.atomic import AtomicOperations
 
 from .api.views_base import DetailViewBase, ListViewBase
 from .models.schemas import (
@@ -134,5 +135,8 @@ def add_routes(app: FastAPI) -> List[Dict[str, Any]]:
         schema_in_post=ComputerInSchema,
     )
 
+    atomic = AtomicOperations()
+
     app.include_router(router, prefix="")
+    app.include_router(atomic.router, prefix="")
     return tags

fastapi_jsonapi/api.py

@@ -40,7 +40,12 @@
 
 
 class RoutersJSONAPI:
-    """API Router interface for JSON API endpoints in web-services."""
+    """
+    API Router interface for JSON API endpoints in web-services.
+    """
+
+    # xxx: store in app, not in routers!
+    all_jsonapi_routers: Dict[str, "RoutersJSONAPI"] = {}
 
     def __init__(
         self,
@@ -96,6 +101,11 @@ def __init__(
         self.model: Type[TypeModel] = model
         self.schema_detail = schema
 
+        if self._type in self.all_jsonapi_routers:
+            msg = f"Resource type {self._type!r} already registered"
+            raise ValueError(msg)
+        self.all_jsonapi_routers[self._type] = self
+
         self.pagination_default_size: Optional[int] = pagination_default_size
         self.pagination_default_number: Optional[int] = pagination_default_number
         self.pagination_default_offset: Optional[int] = pagination_default_offset
@@ -107,8 +117,8 @@ def __init__(
             schema_in_post=schema_in_post,
             schema_in_patch=schema_in_patch,
         )
-        self._schema_in_post = dto.schema_in_post
-        self._schema_in_patch = dto.schema_in_patch
+        self.schema_in_post = dto.schema_in_post
+        self.schema_in_patch = dto.schema_in_patch
         self.detail_response_schema = dto.detail_response_schema
         self.list_response_schema = dto.list_response_schema
 
@@ -394,6 +404,30 @@ def _update_method_config_and_get_dependency_params(
 
         return self._create_dependency_params_from_pydantic_model(method_config.dependencies)
 
+    def get_method_config_for_create(self):
+        method_config = self._update_method_config(
+            view=self.list_view_resource,
+            method=HTTPMethod.POST,
+        )
+        return method_config
+
+    def prepare_dependencies_handler_signature(
+        self,
+        custom_handler: Callable[..., Any],
+        method_config: HTTPMethodConfig,
+    ) -> Signature:
+        sig = signature(custom_handler)
+
+        additional_dependency_params = []
+        if method_config.dependencies is not None:
+            additional_dependency_params = self._create_dependency_params_from_pydantic_model(
+                model_class=method_config.dependencies,
+            )
+
+        params, tail_params = self._get_separated_params(sig)
+
+        return sig.replace(parameters=params + list(additional_dependency_params) + tail_params)
+
     def _create_get_resource_list_view(self):
         """
         Create wrapper for GET list (get objects list)
@@ -425,7 +459,7 @@ def _create_post_resource_list_view(self):
 
         :return:
         """
-        schema_in = self._schema_in_post
+        schema_in = self.schema_in_post
 
         async def wrapper(request: Request, data_create: schema_in, **extra_view_deps):
             resource = self.list_view_resource(
@@ -515,7 +549,7 @@ def _create_patch_resource_detail_view(self):
 
         :return:
         """
-        schema_in = self._schema_in_patch
+        schema_in = self.schema_in_patch
 
         async def wrapper(
             request: Request,

fastapi_jsonapi/atomic/__init__.py

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

fastapi_jsonapi/atomic/atomic.py

@@ -0,0 +1,142 @@
+from dataclasses import dataclass
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    List,
+    Literal,
+    Optional,
+    Type,
+    Union,
+)
+
+from fastapi import APIRouter, Request
+
+from fastapi_jsonapi import RoutersJSONAPI
+from fastapi_jsonapi.atomic.schemas import (
+    AtomicOperationRequest,
+    AtomicResultResponse,
+    OperationItemInSchema,
+    OperationRelationshipSchema,
+)
+from fastapi_jsonapi.utils.dependency_helper import DependencyHelper
+from fastapi_jsonapi.views.utils import HTTPMethodConfig
+from fastapi_jsonapi.views.view_base import ViewBase
+
+if TYPE_CHECKING:
+    from fastapi_jsonapi.data_layers.base import BaseDataLayer
+    from fastapi_jsonapi.views.list_view import ListViewBase
+
+
+@dataclass
+class PreparedOperation:
+    action: Literal["add", "update", "remove"]
+    data_layer: "BaseDataLayer"
+    view: "ViewBase"
+    jsonapi: RoutersJSONAPI
+    data: Union[
+        # from biggest to smallest!
+        # any object creation
+        OperationItemInSchema,
+        # to-many relationship
+        List[OperationRelationshipSchema],
+        # to-one relationship
+        OperationRelationshipSchema,
+        # not required
+        None,
+    ] = None
+
+
+class AtomicOperations:
+    def __init__(
+        self,
+        url_path: str = "/operations",
+        router: Optional[APIRouter] = None,
+    ):
+        self.router = router or APIRouter(tags=["Atomic Operations"])
+        self.url_path = url_path
+        self._register_view()
+
+    async def handle_view_dependencies(
+        self,
+        request: Request,
+        jsonapi: RoutersJSONAPI,
+    ) -> Dict[str, Any]:
+        method_config: HTTPMethodConfig = jsonapi.get_method_config_for_create()
+
+        def handle_dependencies(**dep_kwargs):
+            return dep_kwargs
+
+        handle_dependencies.__signature__ = jsonapi.prepare_dependencies_handler_signature(
+            custom_handler=handle_dependencies,
+            method_config=method_config,
+        )
+
+        dependencies_result: Dict[str, Any] = await DependencyHelper(request=request).run(handle_dependencies)
+        return dependencies_result
+
+    async def view_atomic(
+        self,
+        request: Request,
+        operations_request: AtomicOperationRequest,
+    ):
+        prepared_operations: List[PreparedOperation] = []
+
+        for operation in operations_request.operations:
+            jsonapi = RoutersJSONAPI.all_jsonapi_routers[operation.data.type]
+            view_cls: Type["ViewBase"] = jsonapi.detail_view_resource
+            if operation.op == "add":
+                view_cls = jsonapi.list_view_resource
+            view = view_cls(request=request, jsonapi=jsonapi)
+            dependencies_result: Dict[str, Any] = await self.handle_view_dependencies(
+                request=request,
+                jsonapi=jsonapi,
+            )
+            dl: "BaseDataLayer" = await view.get_data_layer(dependencies_result)
+
+            one_operation = PreparedOperation(
+                action=operation.op,
+                data_layer=dl,
+                view=view,
+                jsonapi=jsonapi,
+                data=operation.data,
+            )
+            prepared_operations.append(one_operation)
+
+        results = []
+
+        for operation in prepared_operations:
+            dl = operation.data_layer
+            if operation.action == "add":
+                data = operation.jsonapi.schema_in_post(data=operation.data)
+                created_object = await dl.create_object(
+                    data_create=data.data,
+                    view_kwargs={},
+                )
+                # assert isinstance(operation.view, ListViewBase)
+                view: "ListViewBase" = operation.view
+                response = await view.response_for_created_object(
+                    dl=operation.data_layer,
+                    created_object=created_object,
+                )
+                results.append({"data": response.data})
+            elif operation.action == "update":
+                # TODO
+                data = operation.jsonapi.schema_in_patch(data=operation.data)
+            elif operation.action == "remove":
+                pass
+            else:
+                msg = f"unknown action {operation.action!r}"
+                raise ValueError(msg)
+
+        return {"atomic:results": results}
+
+    def _register_view(self):
+        self.router.add_api_route(
+            path=self.url_path,
+            endpoint=self.view_atomic,
+            response_model=AtomicResultResponse,
+            methods=["Post"],
+            summary="Atomic operations",
+            description="""[https://jsonapi.org/ext/atomic/](https://jsonapi.org/ext/atomic/)""",
+        )