Added dynamic annotation for API parameters

Author: Znbiz

examples/api_for_sqlalchemy/api/user.py

@@ -9,7 +9,6 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.sql import Select
 from tortoise.exceptions import DoesNotExist
-from tortoise.queryset import QuerySet
 
 from examples.api_for_sqlalchemy.extensions.sqlalchemy import Connector
 from examples.api_for_sqlalchemy.helpers.factories.meta_base import FactoryUseMode
@@ -20,7 +19,6 @@
 from examples.api_for_sqlalchemy.models.pydantic.user import UserInSchema
 from examples.api_for_sqlalchemy.models.sqlalchemy import User
 from fastapi_rest_jsonapi import SqlalchemyEngine
-
 from fastapi_rest_jsonapi.exceptions import (
     BadRequest,
     HTTPException,

fastapi_rest_jsonapi/api.py

@@ -6,14 +6,14 @@
     List,
     Optional,
     Type,
-    Union,
+    Union, TypeVar,
 )
 
 import pydantic
 from fastapi import APIRouter
 from pydantic import BaseModel, Field
 
-from fastapi_rest_jsonapi.data_layers.data_typing import TypeModel, TypeSchema
+from fastapi_rest_jsonapi.data_layers.data_typing import TypeModel
 from fastapi_rest_jsonapi.data_layers.orm import DBORMType
 from fastapi_rest_jsonapi.exceptions import ExceptionResponseSchema
 from fastapi_rest_jsonapi.methods import (
@@ -28,13 +28,16 @@
 
 JSON_API_RESPONSE_TYPE = Optional[Dict[Union[int, str], Dict[str, Any]]]
 
+TypeAPIRouter = TypeVar("TypeAPIRouter", bound=APIRouter)
+TypeSchema = TypeVar("TypeSchema", bound=BaseModel)
+
 
 class RoutersJSONAPI:
     """API Router interface for JSON API endpoints in web-services."""
 
     def __init__(  # noqa: WPS211
         self,
-        routers: APIRouter,
+        routers: TypeAPIRouter,
         path: Union[str, List[str]],
         tags: List[str],
         class_detail: Any,

fastapi_rest_jsonapi/methods.py

@@ -23,10 +23,7 @@
 from fastapi_rest_jsonapi.data_layers.tortoise_orm_engine import TortoiseORMEngine
 from fastapi_rest_jsonapi.exceptions.json_api import UnsupportedFeatureORM
 from fastapi_rest_jsonapi.querystring import QueryStringManager
-from fastapi_rest_jsonapi.signature import (
-    is_necessary_request,
-    update_signature,
-)
+from fastapi_rest_jsonapi.signature import update_signature
 
 
 def get_detail_jsonapi(
@@ -41,13 +38,16 @@ def get_detail_jsonapi(
     def inner(func: Callable) -> Callable:
         async def wrapper(request: Request, obj_id: int, **kwargs):
             query_params = QueryStringManager(request=request, schema=schema)
-            data_dict: dict = dict(query_params=query_params, obj_id=obj_id)
-            if is_necessary_request(func):
-                data_dict["request"] = request
-
-            params_function = OrderedDict(signature(func).parameters)
-            data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in params_function})
-            data_dict = {i_k: i_v for i_k, i_v in data_dict.items() if i_k in params_function}
+            data_dict = {"obj_id": obj_id}
+            func_signature = signature(func).parameters
+            for i_name, i_type in OrderedDict(func_signature).items():
+                if i_type.annotation is Request:
+                    data_dict[i_name] = request
+                elif i_type.annotation is QueryStringManager:
+                    data_dict[i_name] = query_params
+
+            data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in func_signature})
+            data_dict = {i_k: i_v for i_k, i_v in data_dict.items() if i_k in func_signature}
             data_schema: Any = await func(**data_dict)
             return schema_resp(
                 data={
@@ -86,13 +86,18 @@ def patch_detail_jsonapi(
     def inner(func: Callable) -> Callable:
         async def wrapper(request: Request, obj_id: int, data: schema_in, **kwargs):  # type: ignore
             query_params = QueryStringManager(request=request, schema=schema)
-            data_dict: dict = dict(query_params=query_params, obj_id=obj_id, data=getattr(data, "attributes", data))
-            if is_necessary_request(func):
-                data_dict["request"] = request
-
-            params_function = OrderedDict(signature(func).parameters)
-            data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in params_function})
-            data_dict = {i_k: i_v for i_k, i_v in data_dict.items() if i_k in params_function}
+            data_dict = {"obj_id": obj_id}
+            func_signature = signature(func).parameters
+            for i_name, i_type in OrderedDict(func_signature).items():
+                if i_type.annotation is schema_in.__fields__["attributes"].type_:
+                    data_dict[i_name] = getattr(data, 'attributes', data)
+                elif i_type.annotation is Request:
+                    data_dict[i_name] = request
+                elif i_type.annotation is QueryStringManager:
+                    data_dict[i_name] = query_params
+
+            data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in func_signature})
+            data_dict = {i_k: i_v for i_k, i_v in data_dict.items() if i_k in func_signature}
             data_schema: Any = await func(**data_dict)
             return schema_resp(
                 data={
@@ -122,13 +127,16 @@ def delete_detail_jsonapi(
     def inner(func: Callable) -> Callable:
         async def wrapper(request: Request, obj_id: int, **kwargs):  # type: ignore
             query_params = QueryStringManager(request=request, schema=schema)
-            data_dict: dict = dict(query_params=query_params, obj_id=obj_id)
-            if is_necessary_request(func):
-                data_dict["request"] = request
-
-            params_function = OrderedDict(signature(func).parameters)
-            data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in params_function})
-            data_dict = {i_k: i_v for i_k, i_v in data_dict.items() if i_k in params_function}
+            data_dict = {"obj_id": obj_id}
+            func_signature = signature(func).parameters
+            for i_name, i_type in OrderedDict(func_signature).items():
+                if i_type.annotation is Request:
+                    data_dict[i_name] = request
+                elif i_type.annotation is QueryStringManager:
+                    data_dict[i_name] = query_params
+
+            data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in func_signature})
+            data_dict = {i_k: i_v for i_k, i_v in data_dict.items() if i_k in func_signature}
             await func(**data_dict)
             return Response(status_code=status.HTTP_204_NO_CONTENT)
 
@@ -163,9 +171,13 @@ async def wrapper(
             **kwargs,
         ):
             query_params = QueryStringManager(request=request, schema=schema)
-            data_dict: dict = dict(query_params=query_params)
-            if is_necessary_request(func):
-                data_dict["request"] = request
+            data_dict = {}
+            func_signature = signature(func).parameters
+            for i_name, i_type in OrderedDict(func_signature).items():
+                if i_type.annotation is Request:
+                    data_dict[i_name] = request
+                elif i_type.annotation is QueryStringManager:
+                    data_dict[i_name] = query_params
 
             params_function = OrderedDict(signature(func).parameters)
             data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in params_function})
@@ -238,22 +250,21 @@ async def wrapper(
             **kwargs,
         ):
             query_params = QueryStringManager(request=request, schema=schema)
-            data = {
-                i_name: query_params
-                for i_name, i_param in OrderedDict(signature(func).parameters).items()
-                if i_param.annotation is QueryStringManager
-            }
-            if is_necessary_request(func):
-                data["request"] = request
-
-            params_function = OrderedDict(signature(func).parameters)
-            data.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in params_function})
-            data = {i_k: i_v for i_k, i_v in data.items() if i_k in params_function}
-            query = await func(**data)
+            data_dict = {}
+            func_signature = signature(func).parameters
+            for i_name, i_type in OrderedDict(func_signature).items():
+                if i_type.annotation is Request:
+                    data_dict[i_name] = request
+                elif i_type.annotation is QueryStringManager:
+                    data_dict[i_name] = query_params
+
+            data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in func_signature})
+            data_dict = {i_k: i_v for i_k, i_v in data_dict.items() if i_k in func_signature}
+            query = await func(**data_dict)
 
             if engine is DBORMType.sqlalchemy:
                 # Для SQLAlchemy нужно указывать session, для Tortoise достаточно модели
-                session_list = [i_v for i_k, i_v in params_function.items() if isinstance(i_v, AsyncSession)]
+                session_list = [i_v for i_k, i_v in func_signature.items() if isinstance(i_v, AsyncSession)]
                 session: Optional[AsyncSession] = session_list and session_list[0] or None
             else:
                 session = None
@@ -303,11 +314,17 @@ def post_list_jsonapi(
     def inner(func: Callable) -> Callable:
         async def wrapper(request: Request, data: schema_in, **kwargs):  # type: ignore
             query_params = QueryStringManager(request=request, schema=schema)
-            data_dict: dict = dict(query_params=query_params, data=getattr(data, 'attributes', data))
-            if is_necessary_request(func):
-                data_dict["request"] = request
-
-            params_function = OrderedDict(signature(func).parameters)
+            data_dict = {}
+            func_signature = signature(func).parameters
+            for i_name, i_type in OrderedDict(func_signature).items():
+                if i_type.annotation is schema_in.__fields__["attributes"].type_:
+                    data_dict[i_name] = getattr(data, 'attributes', data)
+                elif i_type.annotation is Request:
+                    data_dict[i_name] = request
+                elif i_type.annotation is QueryStringManager:
+                    data_dict[i_name] = query_params
+
+            params_function = OrderedDict(func_signature)
             data_dict.update({i_k: i_v for i_k, i_v in kwargs.items() if i_k in params_function})
             data_dict = {i_k: i_v for i_k, i_v in data_dict.items() if i_k in params_function}
             data_pydantic: Any = await func(**data_dict)

fastapi_rest_jsonapi/schema.py

@@ -34,7 +34,7 @@ class BasePostJSONAPISchema(BaseJSONAPIItemSchema):
 class BaseJSONAPIObjectSchema(BaseJSONAPIItemSchema):
     """Base JSON:API object schema."""
 
-    id: Union[uuid.UUID, int] = Field(description="ID объекта")
+    id: Union[int, uuid.UUID, str] = Field(description="ID объекта")
 
 
 class BasePatchJSONAPISchema(BaseJSONAPIObjectSchema):

fastapi_rest_jsonapi/signature.py

@@ -16,17 +16,11 @@
 
 from fastapi import Query
 from pydantic import BaseModel
+from starlette.requests import Request
 
 from fastapi_rest_jsonapi.querystring import QueryStringManager
 
 
-def is_necessary_request(func: Callable) -> bool:
-    """Check request in parameters."""
-    sig = signature(func)
-    params_dict = OrderedDict(sig.parameters)
-    return "request" in params_dict
-
-
 def update_signature(
         sig: Signature,
         schema: Optional[Type[BaseModel]] = None,
@@ -39,8 +33,28 @@ def update_signature(
     :params schema: которую нужно вставить в сигнатуру.
     :params other: список параметров из начальной функции.
     """
-    other: OrderedDict[str, Parameter] = other or {}
     params_dict = OrderedDict(sig.parameters)
+
+    other_: OrderedDict[str, Parameter] = (other or {}).copy()
+    for i_k, i_v in (other or {}).items():
+        for j_k, j_v in params_dict.items():
+            try:
+                if i_v.annotation is j_v.annotation.__fields__["attributes"].type_:
+                    other_.pop(i_k)
+            except Exception:
+                pass
+            try:
+                if i_v.annotation is Request:
+                    other_.pop(i_k)
+            except Exception:
+                pass
+            try:
+                if i_v.annotation is QueryStringManager:
+                    other_.pop(i_k)
+            except Exception:
+                pass
+    other = other_
+
     params_dict.pop("kwargs", None)
     params_dict.pop("cls", None)
     params_no_default = [