Fastapi Backend: Part 3 (Service Layer)

This is the logs for building service layer

Agenda:

build CRUD service for books
consume the Book model that we built in DB Layer
create DTOs to validate the data from the user
create a common standard response template for all the services
create a custom exception handler for the whole app
configure logging for the whole app

Common standard response schema

src/core/schemas.pyPython

			
from typing import Generic, TypeVar
from pydantic import BaseModel
# Create a type variable that can represent any data schema
T = TypeVar("T")
class StandardResponse(BaseModel, Generic[T]):
    message: str
    data: T | None = None

		

Description:

Most of our app’s response will be of this type
there will be a message
and “data” which can be of any type that we want
if we do not pass anything, then it will be None as default value

Schemas specific to books module

src/features/books/schemas.pyPython

			
import uuid
from typing import Annotated
from pydantic import BaseModel, BeforeValidator, ConfigDict
class CreateBookType(BaseModel):
    title: str
    author: str
AutoStringUUID = Annotated[
    str, BeforeValidator(lambda v: str(v) if isinstance(v, uuid.UUID) else v)
]
class BookResponse(CreateBookType):
    uid: AutoStringUUID
    # makes the response to access the raw db data using dot notation
    # db returns class objects, due to which pydantic cannot access the keys using "[]", which is the default behaviour
    # making the following, tells pydantic to use dot notation to access the data
    model_config = ConfigDict(from_attributes=True)

		

Description:

BookResponse:
- this will be the data type which will be send as a response to the frontend for a book
- this pattern can be really helpful, when we want certain data to be restricted or transformed before sending to the frontend. For example, in case of user, we can restrict password to be sent or get the role or other profile data as a single entity without deeply nested objects
- we were storing uuid as a string in DB but whenever we want to access the uuid using model, the model auto converts to python’s uuid so that we can do any operation using python native uuid. But, we want the same thing to happen in the response as well, where “BeforeValidator” is useful.
- BeforeValidator checks if uuid, present in the data, then it will convert it to string
- the “lambda” delays the execution, making the conversion to happen when the BookResponse is used rather than when it is initialized or at the app’s startup

Books Service:

src/features/books/service.pyPython

			
from sqlmodel import desc, select
from sqlmodel.ext.asyncio.session import AsyncSession
from src.features.books.models import Book
class BookService:
    async def get_all_books(self, session: AsyncSession):
        statement = select(Book).order_by(desc(Book.created_at))
        result = await session.exec(statement)
        return result.all()

		

Description:

a session will be passed down as an argument whihc will be used to fetch all the books list
limit() is encouraged to be used, I will be using it as well

Books routes

src/routes/v1/books_routes.pyPython

			
from src.core.schemas import StandardResponse
from src.features.books.service import book_service
from src.db.session import AsyncSessionDep
from src.features.books.schemas import BookResponse
router = APIRouter(prefix="/books", tags=["Books V1"])
@router.get("/", response_model=StandardResponse[list[BookResponse]])
async def get_all_books(session: AsyncSessionDep):
    books = await book_service.get_all_books(session=session)
    return {"message": "Books fetched succesfully", "data": books}

		

Description:

Session will be injected in this route as DI.
In the response_model, I have used the StandardResponse type which will take a list of BookResponse. Hence, the return dict response replicates it.

Data validation using DTO

Python

			
# src/features/books/schemas.py
class CreateBookType(BaseModel):
    title: str
    author: str
# src/features/books/service.py
class BookService:
    async def create_book(self, book_data: CreateBookType, session: AsyncSession):
        book_data_dict = book_data.model_dump()
        new_book = Book(**book_data_dict)
        session.add(new_book)
        await session.commit()

		

Description:

CreateBookType will act as DTO to validate the data received from user. If “title” or “author” is not present or is of not the string type, the app will throw error

Logging configuration for the app:

I wanted to configure logging throughout the app such that I can log my data anywhere in the app specific to the part where I am logging and all the details so that I can get the idea where it is logged so that I can debug.

src/core/configs/app_config.pyPython

			
def setup_logging():
    logging.basicConfig(
        level=logging.INFO,  # Make INFO, WARNING, and ERROR messages visible
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )
    logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
@asynccontextmanager
async def life_span(app: FastAPI):
    print("server is starting...")
    setup_logging()
    await init_db()
    await ping_auth_db()
    yield
    print("server is stopped")
    await close_db()
    await close_auth_db()

		

Description:

I configured it at the app config and attached it to the app’s lifespan
By default, FastAPI has disabled info level warning, which I have enabled it here
Also, enabled the sqlalchemy engine’s info level

Custom exception:

I wanted to make a centralized section where I can get all the exceptions raised through the app and I can customize according to my need and then send to the user. Hence, I built this:

src/core/exceptions.pyPython

			
import logging
from typing import cast
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import IntegrityError
# from starlette.exception_handlers import http_exception as starlette_http_handler
# from starlette.exceptions import HTTPException as StarletteHTTPException
# Add the correct Starlette internal type for standard 500 exceptions
from starlette.responses import JSONResponse
from starlette.types import ExceptionHandler
logger = logging.getLogger("api_global_logger")
def register_exception_handlers(app: FastAPI) -> None:
    # Register all the global api error interceptors
    # @app.exception_handler(HTTPException)
    async def custom_http_handler(req: Request, exc: HTTPException):
        logger.warning(
            f"⚠️ API HTTP Warning on {req.method} {req.url.path} | Status: {exc.status_code} | Reason: {exc.detail}"
        )
        return await http_exception_handler(req, exc)
    # 2. Catch Frontend Payload Validation Mistakes (422)
    # @app.exception_handler(RequestValidationError)
    async def custom_validation_handler(request: Request, exc: RequestValidationError):
        logger.warning(
            f"🔍 Input Schema Validation Failure on {request.method} {request.url.path} | Errors: {exc.errors()}"
        )
        return await request_validation_exception_handler(request, exc)
    async def custom_integrity_handler(request: Request, exc: IntegrityError):
        logger.exception(
            f"💥 Duplicate/Conflict data on {request.method} {request.url.path} | Error: {str(exc)}"
        )
        # Manually return a clean JSON response to mask internal infrastructure secrets
        return JSONResponse(
            status_code=status.HTTP_409_CONFLICT,
            content={"detail": "A record with provided details already exists"},
        )
    # 3. Catch Severe Internal System Crashes (500)
    # @app.exception_handler(Exception)
    async def custom_global_system_handler(request: Request, exc: Exception):
        logger.exception(
            f"💥 SYSTEM CRASH on {request.method} {request.url.path} | Error: {str(exc)}"
        )
        # fallback_exc = StarletteHTTPException(status_code=500, detail="Internal Server Error")
        # return await starlette_http_handler(request, fallback_exc)
        # Manually return a clean JSON response to mask internal infrastructure secrets
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={
                "detail": "An unexpected server error occurred. Please check logs."
            },
        )
    app.add_exception_handler(
        HTTPException, cast(ExceptionHandler, custom_http_handler)
    )
    app.add_exception_handler(
        RequestValidationError, cast(ExceptionHandler, custom_validation_handler)
    )
    app.add_exception_handler(
        IntegrityError, cast(ExceptionHandler, custom_integrity_handler)
    )
    app.add_exception_handler(Exception, custom_global_system_handler)

		

Description:

I have handled different type of exceptions here
Some specific exceptions are HTTPException, RequestValidationError, IntegrityError
If any other exception occurs, it will be handled at the last one called “Exception”
We can add more exceptions and for each exception we can have custom response to be sent to the user.
I wanted a standard response type as well, like we had for the response for services. This can be integrated easily in frontend so that I can know which key has the error message

So this can be implemented for user service as well or any other service. In the next part I will be implementing authentication.

Fastapi Backend: Part 3 (Service Layer)

Share this:

Leave a comment Cancel reply