feat: add base templates for contextvards, exceptions, responses and …

mypy.ini

@@ -7,3 +7,16 @@ ignore_errors = True
 
 [mypy-src.shared.environment]
 ignore_errors = True
+
+[mypy-src.shared.base_error_internal]
+ignore_errors = True
+
+[mypy-src.shared.base_contextvars]
+ignore_errors = True
+
+[mypy-src.shared.base_responses]
+ignore_errors = True
+
+[mypy-src.shared.base_exceptions]
+ignore_errors = True
+

src/api/v1/endpoints.py

@@ -1,65 +1,52 @@
-from fastapi import HTTPException
-from api.v1.schema import Book, BookCreate
+from api.v1.schema import BookCreateSchema
 from loguru import logger
-
-from fastapi import (
-    APIRouter,
+from shared.base_responses import create_response_for_fast_api, EnvelopeResponse
+from fastapi import APIRouter, Response
+from uuid import UUID
+from api.v1.services import (
+    BooksListService,
+    BookCreateService,
+    BookRetrieveService,
+    BookUpdateService,
+    BookDeleteService,
 )
 
-router = APIRouter(prefix="/books",tags=["Books"])
+router = APIRouter(prefix="/books", tags=["Books"])
 
 
-# Simulated database
-books_db: list[Book] = []
-counter_id = 0
+@router.get("", response_model=EnvelopeResponse)
+async def get_books() -> EnvelopeResponse:
+    logger.info("Retrieving all books")
+    books = BooksListService.list()
+    return create_response_for_fast_api(data=books)
 
 
-# Get all books
-@router.get("", response_model=list[Book])
-async def get_books() -> list[Book]:
-    logger.info("Retrieving all books")
-    return books_db
+@router.post("", response_model=EnvelopeResponse)
+async def create_book(book: BookCreateSchema) -> EnvelopeResponse:
+    logger.info("Creating new book")
+    new_book = BookCreateService.create(book)
+    logger.info(f"Book created with ID: {new_book.id}")
+    return create_response_for_fast_api(data=new_book, status_code_http=201)
 
-@router.post("", response_model=Book, status_code=201)
-async def create_book(book: BookCreate) -> Book:
-    global counter_id
-    counter_id += 1
-    new_book = Book(id=counter_id, **book.model_dump())
-    books_db.append(new_book)
-    logger.info(f"Created new book with ID: {counter_id}")
-    return new_book
 
-# Get book by ID
-@router.get("/{book_id}", response_model=Book)
-async def get_book(book_id: int) -> Book:
+@router.get("/{book_id}", response_model=EnvelopeResponse)
+async def get_book(book_id: UUID) -> EnvelopeResponse:
     logger.info(f"Retrieving book with ID: {book_id}")
-    for book in books_db:
-        if book.id == book_id:
-            return book
-    logger.error(f"Book with ID {book_id} not found")
-    raise HTTPException(status_code=404, detail="Book not found")
+    book = BookRetrieveService.retrieve(book_id)
+    return create_response_for_fast_api(data=book)
+
+
+@router.put("/{book_id}", response_model=EnvelopeResponse)
+async def update_book(book_id: UUID, updated: BookCreateSchema) -> EnvelopeResponse:
+    logger.info(f"Updating book with ID: {book_id}")
+    updated_book = BookUpdateService.update(book_id, updated)
+    logger.info(f"Successfully updated book with ID: {book_id}")
+    return create_response_for_fast_api(data=updated_book)
 
-# Update a book
-@router.put("/{book_id}", response_model=Book)
-async def update_book(book_id: int, updated: BookCreate) -> Book:
-    logger.info(f"Attempting to update book with ID: {book_id}")
-    for i, book in enumerate(books_db):
-        if book.id == book_id:
-            updated_book = Book(id=book_id, **updated.model_dump())
-            books_db[i] = updated_book
-            logger.info(f"Successfully updated book with ID: {book_id}")
-            return updated_book
-    logger.error(f"Book with ID {book_id} not found for update")
-    raise HTTPException(status_code=404, detail="Book not found")
 
-# Delete a book
 @router.delete("/{book_id}", status_code=204)
-async def delete_book(book_id: int) -> None:
+async def delete_book(book_id: UUID) -> None:
     logger.info(f"Attempting to delete book with ID: {book_id}")
-    for i, book in enumerate(books_db):
-        if book.id == book_id:
-            books_db.pop(i)
-            logger.info(f"Successfully deleted book with ID: {book_id}")
-            return
-    logger.error(f"Book with ID {book_id} not found for deletion")
-    raise HTTPException(status_code=404, detail="Book not found")
+    BookDeleteService.delete(book_id)
+    logger.info(f"Successfully deleted book with ID: {book_id}")
+    return Response(status_code=204)

src/api/v1/repositories.py

@@ -0,0 +1,49 @@
+from db.posgresql import get_db_context
+from db.posgresql.models.public import Book
+from api.v1.schema import BookCreateSchema
+from sqlalchemy import select
+
+class BookRepository:
+
+    @staticmethod
+    def get_all() -> tuple[bool, list[Book]]:
+        with get_db_context() as session:
+            books = session.scalars(select(Book)).all()
+            return True, books
+
+    @staticmethod
+    def get_by_id(book_id: int) -> tuple[bool, Book | None]:
+        with get_db_context() as session:
+            book = session.get(Book, book_id)
+            return (True, book) if book else (False, None)
+
+    @staticmethod
+    def create(book_create: BookCreateSchema) -> tuple[bool, Book]:
+        with get_db_context() as session:
+            new_book = Book(**book_create.model_dump())
+            session.add(new_book)
+            session.commit()
+            session.refresh(new_book)
+            return True, new_book
+
+    @staticmethod
+    def update(book_id: int, book_update: BookCreateSchema) -> tuple[bool, Book | None]:
+        with get_db_context() as session:
+            book = session.get(Book, book_id)
+            if not book:
+                return False, None
+            for key, value in book_update.model_dump().items():
+                setattr(book, key, value)
+            session.commit()
+            session.refresh(book)
+            return True, book
+
+    @staticmethod
+    def delete(book_id: int) -> tuple[bool, None]:
+        with get_db_context() as session:
+            book = session.get(Book, book_id)
+            if not book:
+                return False, None
+            session.delete(book)
+            session.commit()
+            return True, None

src/api/v1/schema.py

@@ -1,15 +1,24 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict, field_serializer
+from db.posgresql.models.public import BookType
+from uuid import UUID
 
 # Data model
-class Book(BaseModel):
-    id: int
+class BookSchema(BaseModel):
+    id: UUID
     title: str
     author: str
     year: int
+    type: BookType
 
+    model_config = ConfigDict(validate_by_name=False)
+
+    @field_serializer('id')
+    def serialize_id(self, v: UUID, _info):
+        return str(v)
 
 # Create a new book
-class BookCreate(BaseModel):
+class BookCreateSchema(BaseModel):
     title: str
     author: str
-    year: int
+    year: int
+    type: BookType

src/api/v1/services.py

@@ -0,0 +1,57 @@
+from api.v1.repositories import BookRepository
+from api.v1.schema import BookSchema, BookCreateSchema
+from core.exceptions import BookException
+from uuid import UUID
+
+
+class BooksListService:
+    @staticmethod
+    def list() -> list[BookSchema]:
+        success, list_books = BookRepository.get_all()
+        if not success:
+            raise BookException(message="Failed to retrieve books")
+        return [BookSchema(**book.to_dict()) for book in list_books]
+
+
+class BookCreateService:
+    @staticmethod
+    def create(book_data: BookCreateSchema) -> BookSchema:
+        success, new_book = BookRepository.create(book_data)
+        if not success:
+            raise BookException(message="Failed to create book")
+        return BookSchema(**new_book.to_dict())
+
+
+class BookRetrieveService:
+    @staticmethod
+    def retrieve(book_id: UUID) -> BookSchema:
+        success, book = BookRepository.get_by_id(book_id)
+        if not success:
+            raise BookException(
+                message=f"Book with ID {book_id} not found",
+                data={"payload": {"book_id": str(book_id)}}
+            )
+        return BookSchema(**book.to_dict())
+
+
+class BookUpdateService:
+    @staticmethod
+    def update(book_id: UUID, book_data: BookCreateSchema) -> BookSchema:
+        success, updated_book = BookRepository.update(book_id, book_data)
+        if not success:
+            raise BookException(
+                message=f"Book with ID {book_id} not found for update",
+                data={"payload": {"book_id": str(book_id)}}
+            )
+        return BookSchema(**updated_book.to_dict())
+
+
+class BookDeleteService:
+    @staticmethod
+    def delete(book_id: UUID) -> None:
+        success, _ = BookRepository.delete(book_id)
+        if not success:
+            raise BookException(
+                message=f"Book with ID {book_id} not found for deletion",
+                data={"payload": {"book_id": str(book_id)}}
+            )

src/core/exceptions.py

@@ -0,0 +1,7 @@
+from shared.base_exceptions import BaseApiRestException
+from core.internal_codes import InternalCodesApiBook
+import fastapi
+
+class BookException(BaseApiRestException):
+    GENERAL_ERROR_CODE = InternalCodesApiBook.BOOK_API_ERROR
+    GENERAL_STATUS_CODE_HTTP = fastapi.status.HTTP_400_BAD_REQUEST

src/core/internal_codes.py

@@ -0,0 +1,7 @@
+from shared.base_internal_codes import InternalCodeBase
+
+class InternalCodesApiBook(InternalCodeBase):
+    BOOK_API_ERROR = 1000, "Book API error"
+    BOOK_NOT_FOUND = 1001, "Book not found"
+
+

src/db/mongo/base.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 from uuid import UUID
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ConfigDict, field_serializer
 
 from shared.utils_dates import get_app_current_time
 
@@ -24,11 +24,12 @@ class BaseMongoDocument(BaseModel):
     updated_at: datetime = Field(default_factory=default_mongodb_created_at)
     deleted_at: datetime | None = Field(default=None)
 
-    class Config:
-        allow_population_by_field_name = False
-        json_encoders = {
-            UUID: lambda v: str(v),
-        }
+
+    model_config = ConfigDict(validate_by_name=False)
+
+    @field_serializer('id')
+    def serialize_id(self, v: UUID, _info):
+        return str(v)
 
 
 class MongoAbstractRepository(ABC):

src/db/posgresql/base.py

@@ -24,4 +24,10 @@ def updated_at(cls):
 
     @declared_attr
     def deleted_at(cls):
-        return Column(DateTime, nullable=True)
+        return Column(DateTime, nullable=True)
+
+    def to_dict(self):
+        return {
+            column.name: getattr(self, column.name)
+            for column in self.__table__.columns
+        }

src/db/posgresql/models/public/books.py

@@ -10,7 +10,4 @@ class Book(Base, BaseModel):
     title: str = Column(String, nullable=False)
     author: str = Column(String, nullable=False)
     year: int = Column(Integer, nullable=False)
-    type: BookType = Column(Enum(BookType), nullable=False)
-
-
-
+    type: BookType = Column(Enum(BookType), nullable=False)

src/main.py

@@ -7,13 +7,20 @@
 from api.endpoints import index_router
 from typing import Any
 from core.settings import settings
+from shared.middlewares import (
+    CatcherExceptions,
+    CatcherExceptionsPydantic
+)
+from fastapi.middleware import Middleware
 
 app = FastAPI(
     title=settings.PROJECT.NAME,
     version=settings.PROJECT.VERSION,
     description=settings.PROJECT.DESCRIPTION,
     root_path=settings.ROOT_PATH,
-    middleware=[]
+    middleware=[
+        Middleware(CatcherExceptions)
+    ]
 )
 
 
@@ -34,5 +41,5 @@ def custom_openapi() -> dict[str, Any]:
 app.openapi = custom_openapi # type: ignore
 app.include_router(api_v1_router)
 app.include_router(index_router)
-
+CatcherExceptionsPydantic(app)
 handler = Mangum(app)

src/shared/base_contextvars.py

@@ -0,0 +1,4 @@
+from contextvars import ContextVar
+
+ctx_trace_id = ContextVar("ctx_trace_id", default=None)
+ctx_caller_id = ContextVar("ctx_caller_id", default=None)

src/shared/base_exceptions.py

@@ -0,0 +1,23 @@
+import fastapi
+from loguru import logger
+from shared.base_internal_codes import CommonInternalCode, InternalCode
+
+class BaseApiRestException(Exception):
+    GENERAL_STATUS_CODE_HTTP = fastapi.status.HTTP_400_BAD_REQUEST
+    GENERAL_ERROR_CODE = CommonInternalCode.UNKNOWN
+
+    def __init__(self, 
+                 status_code_http: int = None, 
+                 error_code: InternalCode = None, 
+                 message: str | None = None, 
+                 data: dict[str, any] | None = None):
+        super().__init__(message)
+        self.status_code_http = status_code_http if status_code_http else self.GENERAL_STATUS_CODE_HTTP
+        self.error_code = error_code if error_code else self.GENERAL_ERROR_CODE
+        self.data = data
+        self.message = message
+        logger.warning(self.__str__())
+
+    def __str__(self):
+        return f"[{self.status_code_http}] {self.error_code.description}: {self.message}"
+