Use pydantic v2

Author: febus982

config.py

@@ -1,19 +1,21 @@
 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
+
+# TODO: update this to use the new V2 pydantic-settings
+from pydantic.v1 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 +25,7 @@ class AppConfig(BaseSettings):
                 future=True,
             ),
         ),
-    }
+    )
     ENVIRONMENT: TYPE_ENVIRONMENT = "local"
     DEBUG: bool = False
 

domains/books/_service.py

@@ -29,11 +29,13 @@ async def create_book(self, book: BookData) -> Book:
             some_cpu_intensive_blocking_task, book.dict()
         )
         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: