limit view methods

Author: mahenzon

docs/api_limited_methods_example.rst

@@ -0,0 +1,31 @@
+.. _api_limited_methods_example:
+
+Sometimes you won't need all the CRUD methods.
+For example, you want to create only GET, POST and GET LIST methods,
+so user can't update or delete any items.
+
+
+Set ``methods`` on Routers registration:
+
+.. code-block:: python
+
+    RoutersJSONAPI(
+        router=router,
+        path="/user",
+        tags=["User"],
+        class_detail=UserDetailView,
+        class_list=UserListView,
+        schema=UserSchema,
+        model=User,
+        resource_type="user",
+        methods=[
+            RoutersJSONAPI.Methods.GET_LIST,
+            RoutersJSONAPI.Methods.POST,
+            RoutersJSONAPI.Methods.GET,
+        ],
+    )
+
+
+Full code example:
+
+.. include:: ./api_limited_methods_example.rst

examples/api_limited_methods.py

@@ -0,0 +1,156 @@
+import sys
+from pathlib import Path
+from typing import Any, Dict
+
+import uvicorn
+from fastapi import APIRouter, Depends, FastAPI
+from sqlalchemy import Column, Integer, Text
+from sqlalchemy.engine import make_url
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+from fastapi_jsonapi import RoutersJSONAPI, init
+from fastapi_jsonapi.misc.sqla.generics.base import DetailViewBaseGeneric, ListViewBaseGeneric
+from fastapi_jsonapi.schema_base import BaseModel
+from fastapi_jsonapi.views.utils import HTTPMethod, HTTPMethodConfig
+from fastapi_jsonapi.views.view_base import ViewBase
+
+CURRENT_FILE = Path(__file__).resolve()
+CURRENT_DIR = CURRENT_FILE.parent
+PROJECT_DIR = CURRENT_DIR.parent.parent
+DB_URL = f"sqlite+aiosqlite:///{CURRENT_DIR}/db.sqlite3"
+sys.path.append(str(PROJECT_DIR))
+
+Base = declarative_base()
+
+
+class User(Base):
+    __tablename__ = "users"
+    id = Column(Integer, primary_key=True)
+    name = Column(Text, nullable=True)
+
+
+class UserAttributesBaseSchema(BaseModel):
+    name: str
+
+    class Config:
+        orm_mode = True
+
+
+class UserSchema(UserAttributesBaseSchema):
+    """User base schema."""
+
+
+def async_session() -> sessionmaker:
+    engine = create_async_engine(url=make_url(DB_URL))
+    _async_session = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
+    return _async_session
+
+
+class Connector:
+    @classmethod
+    async def get_session(cls):
+        """
+        Get session as dependency
+
+        :return:
+        """
+        sess = async_session()
+        async with sess() as db_session:  # type: AsyncSession
+            yield db_session
+            await db_session.rollback()
+
+
+async def sqlalchemy_init() -> None:
+    engine = create_async_engine(url=make_url(DB_URL))
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+
+class SessionDependency(BaseModel):
+    session: AsyncSession = Depends(Connector.get_session)
+
+    class Config:
+        arbitrary_types_allowed = True
+
+
+def session_dependency_handler(view: ViewBase, dto: SessionDependency) -> Dict[str, Any]:
+    return {
+        "session": dto.session,
+    }
+
+
+class UserDetailView(DetailViewBaseGeneric):
+    method_dependencies = {
+        HTTPMethod.ALL: HTTPMethodConfig(
+            dependencies=SessionDependency,
+            prepare_data_layer_kwargs=session_dependency_handler,
+        ),
+    }
+
+
+class UserListView(ListViewBaseGeneric):
+    method_dependencies = {
+        HTTPMethod.ALL: HTTPMethodConfig(
+            dependencies=SessionDependency,
+            prepare_data_layer_kwargs=session_dependency_handler,
+        ),
+    }
+
+
+def add_routes(app: FastAPI):
+    tags = [
+        {
+            "name": "User",
+            "description": "",
+        },
+    ]
+
+    router: APIRouter = APIRouter()
+    RoutersJSONAPI(
+        router=router,
+        path="/user",
+        tags=["User"],
+        class_detail=UserDetailView,
+        class_list=UserListView,
+        schema=UserSchema,
+        model=User,
+        resource_type="user",
+        methods=[
+            RoutersJSONAPI.Methods.GET_LIST,
+            RoutersJSONAPI.Methods.POST,
+            RoutersJSONAPI.Methods.GET,
+        ],
+    )
+
+    app.include_router(router, prefix="")
+    return tags
+
+
+def create_app() -> FastAPI:
+    """
+    Create app factory.
+
+    :return: app
+    """
+    app = FastAPI(
+        title="FastAPI app with limited methods",
+        debug=True,
+        openapi_url="/openapi.json",
+        docs_url="/docs",
+    )
+    add_routes(app)
+    app.on_event("startup")(sqlalchemy_init)
+    init(app)
+    return app
+
+
+app = create_app()
+
+if __name__ == "__main__":
+    uvicorn.run(
+        app,
+        host="0.0.0.0",
+        port=8080,
+    )

examples/api_minimal.py

@@ -42,14 +42,6 @@ class UserSchema(UserAttributesBaseSchema):
     """User base schema."""
 
 
-class UserPatchSchema(UserAttributesBaseSchema):
-    """User PATCH schema."""
-
-
-class UserInSchema(UserAttributesBaseSchema):
-    """User input schema."""
-
-
 def async_session() -> sessionmaker:
     engine = create_async_engine(url=make_url(DB_URL))
     _async_session = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
@@ -153,9 +145,7 @@ def create_app() -> FastAPI:
 
 if __name__ == "__main__":
     uvicorn.run(
-        "main:app",
+        app,
         host="0.0.0.0",
         port=8080,
-        reload=True,
-        app_dir=str(CURRENT_DIR),
     )

fastapi_jsonapi/api.py

@@ -1,4 +1,5 @@
 """JSON API router class."""
+from enum import Enum, auto
 from inspect import Parameter, Signature, signature
 from typing import (
     TYPE_CHECKING,
@@ -40,13 +41,24 @@
 not_passed = object()
 
 
+class ViewMethods(str, Enum):
+    GET_LIST = auto()
+    POST = auto()
+    DELETE_LIST = auto()
+    GET = auto()
+    DELETE = auto()
+    PATCH = auto()
+
+
 class RoutersJSONAPI:
     """
     API Router interface for JSON API endpoints in web-services.
     """
 
     # xxx: store in app, not in routers!
     all_jsonapi_routers: Dict[str, "RoutersJSONAPI"] = {}
+    Methods = ViewMethods
+    DEFAULT_METHODS = tuple(str(method) for method in ViewMethods)
 
     def __init__(
         self,
@@ -64,6 +76,7 @@ def __init__(
         pagination_default_number: Optional[int] = 1,
         pagination_default_offset: Optional[int] = None,
         pagination_default_limit: Optional[int] = None,
+        methods: Iterable[str] = (),
     ) -> None:
         """
         Initialize router items.
@@ -101,6 +114,8 @@ def __init__(
         self.schema_list: Type[BaseModel] = schema
         self.model: Type[TypeModel] = model
         self.schema_detail = schema
+        # tuple and not set, so ordering is persisted
+        self.methods = tuple(methods) or self.DEFAULT_METHODS
 
         if self.type_ in self.all_jsonapi_routers:
             msg = f"Resource type {self.type_!r} already registered"
@@ -664,10 +679,19 @@ def _register_views(self, path: str):
         :param path:
         :return:
         """
-        self._register_get_resource_list(path)
-        self._register_post_resource_list(path)
-        self._register_delete_resource_list(path)
-
-        self._register_get_resource_detail(path)
-        self._register_patch_resource_detail(path)
-        self._register_delete_resource_detail(path)
+        methods_map: Dict[Union[str, ViewMethods], Callable[[str], None]] = {
+            ViewMethods.GET_LIST: self._register_get_resource_list,
+            ViewMethods.POST: self._register_post_resource_list,
+            ViewMethods.DELETE_LIST: self._register_delete_resource_list,
+            ViewMethods.GET: self._register_get_resource_detail,
+            ViewMethods.PATCH: self._register_patch_resource_detail,
+            ViewMethods.DELETE: self._register_delete_resource_detail,
+        }
+        # patch for Python < 3.11
+        for key, value in list(methods_map.items()):
+            methods_map[str(key)] = value
+
+        for method in self.methods:
+            # `to str` so Python < 3.11 is supported
+            register = methods_map[str(method)]
+            register(path)