Merge pull request #63 from febus982/pydantic_v2

Use pydantic v2

Author: febus982

config.py

@@ -1,19 +1,19 @@
 import logging
 import os
-from typing import List, Literal
+from typing import Dict, List, Literal
 
 import structlog
 from opentelemetry import trace
-from pydantic import BaseSettings
+from pydantic_settings import BaseSettings
 from sqlalchemy_bind_manager import SQLAlchemyAsyncConfig
 from structlog.typing import Processor
 
 TYPE_ENVIRONMENT = Literal["local", "test", "staging", "production"]
 
 
 class AppConfig(BaseSettings):
-    SQLALCHEMY_CONFIG = {
-        "default": SQLAlchemyAsyncConfig(
+    SQLALCHEMY_CONFIG: Dict[str, SQLAlchemyAsyncConfig] = dict(
+        default=SQLAlchemyAsyncConfig(
             engine_url=f"sqlite+aiosqlite:///{os.path.dirname(os.path.abspath(__file__))}/sqlite.db",
             engine_options=dict(
                 connect_args={
@@ -23,7 +23,7 @@ class AppConfig(BaseSettings):
                 future=True,
             ),
         ),
-    }
+    )
     ENVIRONMENT: TYPE_ENVIRONMENT = "local"
     DEBUG: bool = False
 

domains/books/_dto.py

@@ -7,12 +7,6 @@ class BookData(BaseModel):
     title: str
     author_name: str
 
-    class Config:
-        orm_mode = True
-
 
 class Book(BookData):
     book_id: Union[int, None] = None
-
-    class Config:
-        orm_mode = True

domains/books/_service.py

@@ -26,14 +26,16 @@ async def create_book(self, book: BookData) -> Book:
         # Using processes could be better, but it would bring technical complexity
         # https://anyio.readthedocs.io/en/3.x/subprocesses.html#running-functions-in-worker-processes
         book_data_altered = await to_thread.run_sync(
-            some_cpu_intensive_blocking_task, book.dict()
+            some_cpu_intensive_blocking_task, book.model_dump()
         )
         book_model = BookModel(**book_data_altered)
-        return Book.from_orm(await self.book_repository.save(book_model))
+        return Book.model_validate(
+            await self.book_repository.save(book_model), from_attributes=True
+        )
 
     async def list_books(self) -> Iterable[Book]:
         books = await self.book_repository.find()
-        return [Book.from_orm(x) for x in books]
+        return [Book.model_validate(x, from_attributes=True) for x in books]
 
 
 def some_cpu_intensive_blocking_task(book: dict) -> dict:

grpc_app/servicers/books.py

@@ -14,6 +14,7 @@ async def ListBooks(
         book_service = BookService()
         return books_messages.ListBooksResponse(
             books=[
-                books_messages.Book(**x.dict()) for x in await book_service.list_books()
+                books_messages.Book(**x.model_dump())
+                for x in await book_service.list_books()
             ]
         )

http_app/routes/api/books.py

@@ -1,5 +1,5 @@
 from fastapi import APIRouter, status
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 from domains.books import Book, BookData, BookService
 
@@ -9,28 +9,29 @@
 
 class CreateBookResponse(BaseModel):
     book: Book
-
-    class Config:
-        schema_extra = {
+    model_config = ConfigDict(
+        json_schema_extra={
             "example": {
                 "title": "The Hitchhiker's Guide to the Galaxy",
                 "author_name": "Douglas Adams",
                 "book_id": 123,
             }
         }
+    )
 
 
 class CreateBookRequest(BaseModel):
     title: str
     author_name: str
 
-    class Config:
-        schema_extra = {
+    model_config = ConfigDict(
+        json_schema_extra={
             "example": {
                 "title": "The Hitchhiker's Guide to the Galaxy",
                 "author_name": "Douglas Adams",
             }
         }
+    )
 
 
 """
@@ -49,7 +50,9 @@ async def create_book(
     data: CreateBookRequest,
 ) -> CreateBookResponse:
     book_service = BookService()
-    created_book = await book_service.create_book(book=BookData(**data.dict()))
+    created_book = await book_service.create_book(
+        book=BookData.model_validate(data, from_attributes=True)
+    )
     return CreateBookResponse(book=created_book)
 
 
@@ -59,5 +62,7 @@ async def create_book_v2(
     some_optional_query_param: bool = False,
 ) -> CreateBookResponse:
     book_service = BookService()
-    created_book = await book_service.create_book(book=BookData(**data.dict()))
+    created_book = await book_service.create_book(
+        book=BookData.model_validate(data, from_attributes=True)
+    )
     return CreateBookResponse(book=created_book)

http_app/routes/graphql/resolvers.py

@@ -4,4 +4,4 @@
 async def list_books():
     book_service = BookService()
     books = await book_service.list_books()
-    return [Book(**x.dict()) for x in books]
+    return [Book.model_validate(x, from_attributes=True) for x in books]

http_app/routes/ping.py

@@ -1,20 +1,21 @@
 from fastapi import APIRouter
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 router = APIRouter()
 
 
 class PingResponse(BaseModel):
     ping: str
 
-    class Config:
-        schema_extra = {
+    model_config = ConfigDict(
+        json_schema_extra={
             "example": {
                 "ping": "pong!",
             }
         }
+    )
 
 
-@router.get("/ping", response_model=PingResponse)
+@router.get("/ping")
 async def ping() -> PingResponse:
     return PingResponse(ping="pong!")