upgrade to new fastapi and pydantic (WIP)

examples/api_for_tortoise_orm/api/user.py

@@ -48,7 +48,7 @@ async def get_user(cls, user_id: int, query_params: QueryStringManager) -> User:
     @classmethod
     async def get(cls, obj_id: int, query_params: QueryStringManager) -> UserSchema:
         user: User = await cls.get_user(user_id=obj_id, query_params=query_params)
-        return UserSchema.from_orm(user)
+        return UserSchema.model_validate(user)
 
     @classmethod
     async def patch(cls, obj_id: int, data: UserPatchSchema, query_params: QueryStringManager) -> UserSchema:
@@ -64,7 +64,7 @@ async def patch(cls, obj_id: int, data: UserPatchSchema, query_params: QueryStri
         except ObjectNotFound as ex:
             raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail=ex.description)
 
-        user = UserSchema.from_orm(user_obj)
+        user = UserSchema.model_validate(user_obj)
         return user
 
 
@@ -75,22 +75,22 @@ async def get(cls, query_params: QueryStringManager) -> Union[QuerySet, JSONAPIR
         dl = TortoiseDataLayer(query=user_query, schema=UserSchema, model=User)
         count, users_db = await dl.get_collection(qs=query_params)
         total_pages = count // query_params.pagination.size + (count % query_params.pagination.size and 1)
-        users: List[UserSchema] = [UserSchema.from_orm(i_user) for i_user in users_db]
+        users: List[UserSchema] = [UserSchema.model_validate(i_user) for i_user in users_db]
         return JSONAPIResultListSchema(
             meta={"count": count, "totalPages": total_pages},
-            data=[{"id": i_obj.id, "attributes": i_obj.dict(), "type": "user"} for i_obj in users],
+            data=[{"id": i_obj.id, "attributes": i_obj.model_dump(), "type": "user"} for i_obj in users],
         )
 
     @classmethod
     async def post(cls, data: UserInSchema, query_params: QueryStringManager) -> UserSchema:
         try:
             user_obj = await UserFactory.create(
-                data=data.dict(),
+                data=data.model_dump(),
                 mode=FactoryUseMode.production,
                 header=query_params.headers,
             )
         except ErrorCreateUserObject as ex:
             raise BadRequest(ex.description, ex.field)
 
-        user = UserSchema.from_orm(user_obj)
+        user = UserSchema.model_validate(user_obj)
         return user

fastapi_jsonapi/atomic/atomic_handler.py

@@ -46,7 +46,7 @@ async def wrapper(*a, operation: OperationBase, **kw):
             errors_details = {
                 "message": f"Validation error on operation {operation.op_type}",
                 "ref": operation.ref,
-                "data": operation.data.dict(),
+                "data": operation.data.model_dump(),
             }
             if isinstance(ex, ValidationError):
                 errors_details.update(errors=ex.errors())

fastapi_jsonapi/atomic/schemas.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from enum import Enum
-from typing import List, Optional, Union
+from typing import Annotated, List, Optional, Union
 
 from pydantic import BaseModel, Field, root_validator
 from starlette.datastructures import URLPath
@@ -177,10 +177,21 @@ def validate_operation(cls, values: dict):
 
 
 class AtomicOperationRequest(BaseModel):
-    operations: List[AtomicOperation] = Field(
-        alias="atomic:operations",
-        min_items=1,
-    )
+    # operations: conlist(AtomicOperation, min_length=1)
+    # operations: conlist(AtomicOperation, min_length=1)
+    operations: Annotated[
+        List[AtomicOperation],
+        Field(alias="atomic:operations"),
+    ]
+
+    # operations: conlist(AtomicOperation, min_length=1) = Field(
+    #     alias="atomic:operations",
+    # )
+    # operations: List[AtomicOperation] = Field(
+    #     alias="atomic:operations",
+    #     # TODO: how to limit?
+    #     # min_length=1,
+    # )
 
 
 class AtomicResult(BaseModel):
@@ -199,7 +210,15 @@ class AtomicResultResponse(BaseModel):
     https://jsonapi.org/ext/atomic/#auto-id-responses-4
     """
 
-    results: List[AtomicResult] = Field(
-        alias="atomic:results",
-        min_items=1,
-    )
+    results: Annotated[
+        List[AtomicResult],
+        Field(
+            alias="atomic:results",
+            # min_length=1,
+        ),
+    ]
+    # results: List[AtomicResult] = Field(
+    #     alias="atomic:results",
+    #     # TODO: how to limit?
+    #     min_length=1,
+    # )

fastapi_jsonapi/exceptions/base.py

@@ -24,7 +24,7 @@ class ExceptionSchema(BaseModel):
     status: str
     source: Optional[ExceptionSourceSchema] = None
     title: str
-    detail: Any
+    detail: Any = None
 
 
 class ExceptionResponseSchema(BaseModel):

fastapi_jsonapi/schema.py

@@ -14,9 +14,8 @@
 
 from fastapi import FastAPI
 from pydantic import (
-    BaseConfig,
     BaseModel,
-    Extra,
+    ConfigDict,
     Field,
 )
 
@@ -25,12 +24,13 @@
 
 
 class BaseJSONAPIRelationshipSchema(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+
     id: str = Field(..., description="Related object ID")
     type: str = Field(..., description="Type of the related resource object")
 
-    class Config(BaseConfig):
-        extra = Extra.forbid
-
 
 class BaseJSONAPIRelationshipDataToOneSchema(BaseModel):
     data: BaseJSONAPIRelationshipSchema
@@ -56,7 +56,7 @@ class BaseJSONAPIItemInSchema(BaseJSONAPIItemSchema):
 
     attributes: "TypeSchema" = Field(description="Resource object attributes")
     relationships: Optional["TypeSchema"] = Field(None, description="Resource object relationships")
-    id: Optional[str] = Field(description="Resource object ID")
+    id: Optional[str] = Field(None, description="Resource object ID")
 
 
 class BaseJSONAPIDataInSchema(BaseModel):
@@ -72,11 +72,16 @@ class BaseJSONAPIObjectSchema(BaseJSONAPIItemSchema):
 class JSONAPIResultListMetaSchema(BaseModel):
     """JSON:API list meta schema."""
 
-    count: Optional[int]
-    total_pages: Optional[int] = Field(alias="totalPages")
+    model_config = ConfigDict(
+        extra="forbid",
+        populate_by_name=True,
+    )
 
-    class Config:
-        allow_population_by_field_name = True
+    count: Optional[int] = None
+    total_pages: Optional[int] = Field(
+        None,
+        serialization_alias="totalPages",
+    )
 
 
 class JSONAPIDocumentObjectSchema(BaseModel):
@@ -97,7 +102,7 @@ class BaseJSONAPIResultSchema(BaseModel):
     JSON:API Required fields schema
     """
 
-    meta: Optional[JSONAPIResultListMetaSchema] = Field(description="JSON:API metadata")
+    meta: Optional[JSONAPIResultListMetaSchema] = Field(None, description="JSON:API metadata")
     jsonapi: JSONAPIDocumentObjectSchema = JSONAPIDocumentObjectSchema()
 
 

fastapi_jsonapi/schema_base.py

@@ -1,3 +1,5 @@
+from pydantic import ConfigDict
+
 __all__ = (
     "Field",
     "BaseModel",
@@ -43,13 +45,12 @@ class BaseModel(BaseModelGeneric, metaclass=RegistryMeta):
 
 
 class RelationshipInfo(BaseModel):
+    model_config = ConfigDict(
+        frozen=True,
+    )
     resource_type: str
     many: bool = False
     related_view: str = None
     related_view_kwargs: Dict[str, str] = {}
     resource_id_example: str = "1"
     id_field_name: str = "id"
-
-    # TODO: Pydantic V2 use model_config
-    class Config:
-        frozen = True

fastapi_jsonapi/views/utils.py

@@ -2,7 +2,7 @@
 from functools import cache
 from typing import Callable, Coroutine, Optional, Set, Type, Union
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 
 class HTTPMethod(Enum):
@@ -18,11 +18,11 @@ def names() -> Set[str]:
 
 
 class HTTPMethodConfig(BaseModel):
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+    )
     dependencies: Optional[Type[BaseModel]] = None
-    prepare_data_layer_kwargs: Optional[Union[Callable, Coroutine]] = None
-
-    class Config:
-        arbitrary_types_allowed = True
+    prepare_data_layer_kwargs: Optional[Callable] = None
 
     @property
     def handler(self) -> Optional[Union[Callable, Coroutine]]:

fastapi_jsonapi/views/view_base.py

@@ -192,7 +192,10 @@ def _build_list_response(self, items_from_db: List[TypeModel], count: int, total
             includes_schemas=object_schemas.included_schemas_list,
         )
         return list_jsonapi_schema(
-            meta=JSONAPIResultListMetaSchema(count=count, total_pages=total_pages),
+            meta=JSONAPIResultListMetaSchema(
+                count=count,
+                total_pages=total_pages,
+            ),
             data=result_objects,
             **extras,
         )
@@ -259,7 +262,7 @@ def update_related_object(
         if hasattr(parent_included_object, "relationships") and parent_included_object.relationships:
             existing = parent_included_object.relationships or {}
             if isinstance(existing, BaseModel):
-                existing = existing.dict()
+                existing = existing.model_dump()
             new_relationships.update(existing)
         new_relationships.update(
             **{
@@ -417,7 +420,7 @@ def process_db_object(
 
         item_as_schema = object_schemas.object_jsonapi_schema(
             id=self.get_db_item_id(item),
-            attributes=object_schemas.attributes_schema.from_orm(item),
+            attributes=object_schemas.attributes_schema.model_validate(item),
         )
 
         cache_included_objects: Dict[Tuple[str, str], TypeSchema] = {}