create get_data_layer method

Author: mahenzon

fastapi_jsonapi/views/detail_view.py

@@ -1,5 +1,11 @@
 import logging
-from typing import TYPE_CHECKING, TypeVar, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    TypeVar,
+    Union,
+)
 
 from fastapi_jsonapi import BadRequest
 from fastapi_jsonapi.schema import (
@@ -18,12 +24,18 @@
 
 
 class DetailViewBase(ViewBase):
+    async def get_data_layer(
+        self,
+        extra_view_deps: Dict[str, Any],
+    ) -> "BaseDataLayer":
+        return await self.get_data_layer_for_detail(extra_view_deps)
+
     async def handle_get_resource_detail(
         self,
         object_id: Union[int, str],
         **extra_view_deps,
     ):
-        dl: "BaseDataLayer" = await self.get_data_layer_for_detail(extra_view_deps)
+        dl: "BaseDataLayer" = await self.get_data_layer(extra_view_deps)
 
         view_kwargs = {dl.url_id_field: object_id}
         db_object = await dl.get_object(view_kwargs=view_kwargs, qs=self.query_params)
@@ -41,7 +53,7 @@ async def handle_update_resource(
                 detail="obj_id and data.id should be same",
                 pointer="/data/id",
             )
-        dl: "BaseDataLayer" = await self.get_data_layer_for_detail(extra_view_deps)
+        dl: "BaseDataLayer" = await self.get_data_layer(extra_view_deps)
 
         view_kwargs = {dl.url_id_field: obj_id}
         db_object = await dl.get_object(view_kwargs=view_kwargs, qs=self.query_params)
@@ -55,7 +67,7 @@ async def handle_delete_resource(
         obj_id: str,
         **extra_view_deps,
     ):
-        dl: "BaseDataLayer" = await self.get_data_layer_for_detail(extra_view_deps)
+        dl: "BaseDataLayer" = await self.get_data_layer(extra_view_deps)
 
         view_kwargs = {dl.url_id_field: obj_id}
         db_object = await dl.get_object(view_kwargs=view_kwargs, qs=self.query_params)

fastapi_jsonapi/views/list_view.py

@@ -1,5 +1,5 @@
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, Dict
 
 from fastapi_jsonapi.schema import (
     BaseJSONAPIDataInSchema,
@@ -28,8 +28,14 @@ def _calculate_total_pages(self, db_items_count: int) -> int:
 
         return total_pages
 
+    async def get_data_layer(
+        self,
+        extra_view_deps: Dict[str, Any],
+    ) -> "BaseDataLayer":
+        return await self.get_data_layer_for_list(extra_view_deps)
+
     async def handle_get_resource_list(self, **extra_view_deps) -> JSONAPIResultListSchema:
-        dl: "BaseDataLayer" = await self.get_data_layer_for_list(extra_view_deps)
+        dl: "BaseDataLayer" = await self.get_data_layer(extra_view_deps)
         query_params = self.query_params
         count, items_from_db = await dl.get_collection(qs=query_params)
         total_pages = self._calculate_total_pages(count)
@@ -41,7 +47,7 @@ async def handle_post_resource_list(
         data_create: BaseJSONAPIDataInSchema,
         **extra_view_deps,
     ) -> JSONAPIResultDetailSchema:
-        dl: "BaseDataLayer" = await self.get_data_layer_for_list(extra_view_deps)
+        dl: "BaseDataLayer" = await self.get_data_layer(extra_view_deps)
         created_object = await dl.create_object(data_create=data_create.data, view_kwargs={})
         created_object_id = dl.get_object_id(created_object)
 
@@ -51,7 +57,7 @@ async def handle_post_resource_list(
         return self._build_detail_response(db_object)
 
     async def handle_delete_resource_list(self, **extra_view_deps) -> JSONAPIResultListSchema:
-        dl: "BaseDataLayer" = await self.get_data_layer_for_list(extra_view_deps)
+        dl: "BaseDataLayer" = await self.get_data_layer(extra_view_deps)
         query_params = self.query_params
         count, items_from_db = await dl.get_collection(qs=query_params)
         total_pages = self._calculate_total_pages(count)

fastapi_jsonapi/views/view_base.py

@@ -69,6 +69,12 @@ def _get_data_layer(self, schema: Type[BaseModel], **dl_kwargs):
             **dl_kwargs,
         )
 
+    async def get_data_layer(
+        self,
+        extra_view_deps: Dict[str, Any],
+    ) -> BaseDataLayer:
+        raise NotImplementedError
+
     async def get_data_layer_for_detail(
         self,
         extra_view_deps: Dict[str, Any],