Fast API — Repository Pattern and Service Layer

Note

For additional examples, please refer to my complete project available here. This project encompasses more models, user authentication and authorization, as well as unit tests and integration tests.

Tech stack:

• Python
• FastAPI
• Pydantic
• SQLAlchemy
• SQLite

This article does not aim to provide a step-by-step guide on utilizing FastAPI. Hence, it does not delve into every step required, such as elucidating the disparity between a schema and a model class, or how to establish a connection between the application and the database.

Introduction

In many tutorials concerning web technologies in Python (Django, FastAPI, Flask), there’s often a recurring pattern where the entire application logic resides in a single place — views.

For example:

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

However, as our application expands, with more tables and dependencies in the database, this approach becomes less optimal.

It took me a while to realize this, as I observed the same pattern in every Django/FastAPI/Flask guide.

Presently, in my projects, I adopt a combination of the Repository pattern and Service layer.

The Basic configuration of the Fast API project

Let’s start with installing the Fast API library

pip install fastapi[all]

The project structure would resemble the following:

main_folder/
- config/
	- database.py
- models/
	- region.py
	- city.py
- schemas/
	- region_schema.py
	- city_schemas.py
- repository/
	- region_repository.py
	- city_repository.py
- service/
	- region_service.py
	- city_service.ppy
- router/
	- api.py
	- v1/
		- region.py
		- city.py
- utils/
	- init_db.py
__init__.py
main.py

Let’s start by creating config/databse.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine(
    "sqlite:///database.db",
    connect_args={"check_same_thread": False},
)
SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
)
Base = declarative_base()
def get_db():
    """
    Create a database session.
    Yields:
        Session: The database session.
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Now create our models City and Region

City and Region are in one to many relation. City can has assigned one region and one region can be assigned to multiple cities.

models/region.py

import uuid

from sqlalchemy import Column, String
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from config.database import Base


class Region(Base):
    __tablename__ = "regions"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String, nullable=False)
    cities = relationship("City", back_populates="region")

models/city.py

import uuid
from sqlalchemy import Column, String, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from config.database import Base


class City(Base):
    __tablename__ = "cities"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String, nullable=False)
    region_id = Column(UUID(as_uuid=True), ForeignKey("regions.id"))
    region = relationship("Region", back_populates="cities")

To initialize models we need to create function. Let’s create utils/init_db.py file

from config.database import Base, engine
from models.region import Region
from models.city import City


def create_tables():
    """
    Creates all database tables defined in the application.
    """
    Region.metadata.create_all(bind=engine)
    City.metadata.create_all(bind=engine)

Now create schemas

schemas/region_schema.py

from pydantic import BaseModel, UUID4, Field


class RegionInput(BaseModel):
    name: str = Field(min_length=1, max_length=120)


class RegionOutput(BaseModel):
    id: UUID4
    name: str

schemas/city_schema.py

from pydantic import BaseModel, UUID4, Field
from .region_schema import RegionOutput


class CityInput(BaseModel):
    name: str = Field(min_length=1, max_length=120)
    region_id: UUID4


class CityInDb(BaseModel):
    id: UUID4
    name: str
    region_id: UUID4

    class Config:
        orm_mode = True


class CityOutput(BaseModel):
    id: UUID4
    name: str
    region: RegionOutput

That is almost the whole configuration, the last thing we need to do is to create main.py

from fastapi import FastAPI
# from routers.api import router # add this later
from utils.init_db import create_tables


app = FastAPI(
    debug=True,
    title="Tutorial",
)


@app.on_event("startup")
def on_startup() -> None:
    """
    Initializes the database tables when the application starts up.
    """
    create_tables()


# app.include_router(router)

To test if everything works properly use this command to run Fast API application

uvicorn main:app --reload

Repository

The Repository pattern abstracts the data access layer by offering a clean interface for interacting with the underlying data storage, whether it’s a database, external API, or any other data source. It encapsulates the logic for querying, creating, updating, and deleting data entities, thereby promoting the separation of concerns and enhancing the maintainability and testability of the codebase.

Let’s delve into a practical implementation of the Repository pattern in a FastAPI application, utilizing Python’s SQLAlchemy for database ORM (Object-Relational Mapping) and Pydantic for data validation and serialization.

The repository layer consists of classes responsible for database operations. Below is an example of repository classes for handling regions and cities.

The Repository layer should return schema objects, e.g., RegionInput or RegionOutput. However, in some cases, the methods return a model object - Region. This is because all the logic related to the application should be located in the service layer. For instance, when we want to delete the Region object, we pass the object returned from the get_by_id method to the delete method, instead of passing the uuid and checking in the delete method whether the object exists, and then looking for it in the database.

repository/region_repository.py

from sqlalchemy.orm import Session
from models.region import Region
from schemas.region_schema import RegionInput, RegionOutput
from typing import List, Optional, Type
from pydantic import UUID4


class RegionRepository:
    def __init__(self, session: Session):
        self.session = session

    def create(self, data: RegionInput) -> RegionOutput:
        region = Region(**data.model_dump(exclude_none=True))
        self.session.add(region)
        self.session.commit()
        self.session.refresh(region)
        return RegionOutput(id=region.id, name=region.name)

    def get_all(self) -> List[Optional[RegionOutput]]:
        regions = self.session.query(Region).all()
        return [RegionOutput(**region.__dict__) for region in regions]

    def get_region(self, _id: UUID4) -> RegionOutput:
        region = self.session.query(Region).filter_by(id=_id).first()
        return RegionOutput(**region.__dict__)

    def get_by_id(self, _id: UUID4) -> Type[Region]:
        return self.session.query(Region).filter_by(id=_id).first()

    def region_exists_by_id(self, _id: UUID4) -> bool:
        region = self.session.query(Region).filter_by(id=_id).first()
        return region is not None

    def region_exists_by_name(self, name: str) -> bool:
        region = self.session.query(Region).filter_by(name=name).first()
        return region is not None

    def update(self, region: Type[Region], data: RegionInput) -> RegionInput:
        region.name = data.name
        self.session.commit()
        self.session.refresh(region)
        return RegionInput(**region.__dict__)

    def delete(self, region: Type[Region]) -> bool:
        self.session.delete(region)
        self.session.commit()
        return True

repository/city_repository.py

from typing import List, Optional, Type

from models.city import City
from pydantic import UUID4
from schemas.city_schema import CityInput, CityOutput, CityInDb
from schemas.region_schema import RegionOutput
from sqlalchemy.orm import Session


class CityRepository:
    def __init__(self, session: Session):
        self.session = session

    def create(self, data: CityInput) -> CityInDb:
        city = City(**data.model_dump(exclude_none=True))
        self.session.add(city)
        self.session.commit()
        self.session.refresh(city)
        return CityInDb(**city.__dict__)

    def get_all(self) -> List[Optional[CityOutput]]:
        cities = self.session.query(City).all()
        return self._map_city_to_schema_list(cities)

    def get_all_by_region(self, region_id: UUID4) -> List[Optional[CityOutput]]:
        cities = self.session.query(City).filter_by(region_id=region_id).all()
        return self._map_city_to_schema_list(cities)

    def get_by_id(self, _id: UUID4) -> Type[City]:
        return self.session.query(City).filter_by(id=_id).first()

    def city_exists_by_name(self, name: str) -> bool:
        city = self.session.query(City).filter_by(name=name).first()
        return bool(city)

    def city_exists_by_id(self, _id: UUID4) -> bool:
        city = self.session.query(City).filter_by(id=_id).first()
        return bool(city)

    def update(self, city: Type[City], data: CityInput) -> CityInput:
        for key, value in data.model_dump(exclude_none=True).items():
            setattr(city, key, value)
        self.session.commit()
        self.session.refresh(city)
        return CityInput(**city.__dict__)

    def delete(self, city: Type[City]) -> bool:
        self.session.delete(city)
        self.session.commit()
        return True
        
    @staticmethod
    def _map_city_to_schema_list(cities: List[Type[City]]) -> List[CityOutput]:
        return [
            CityOutput(
                id=city.id,
                name=city.name,
                region=RegionOutput(
                    id=city.region.id, name=city.region.name
                )
            )
            for city in cities
        ]

Service

The service layer acts as the intermediary between the API endpoints and the repository layer. It’s responsible for implementing business logic, orchestrating interactions between different repositories, and performing necessary validations or additional operations.

service/region_service.py

from typing import List, Optional

from fastapi import HTTPException
from pydantic import UUID4
from sqlalchemy.orm import Session
from repository.region_repository import RegionRepository
from schemas.region_schema import RegionInput, RegionOutput


class RegionService:
    def __init__(self, session: Session):
        self.repository = RegionRepository(session)

    def create(self, data: RegionInput) -> RegionOutput:
        if self.repository.region_exists_by_name(data.name):
            raise HTTPException(status_code=400, detail="Region already exists")
        return self.repository.create(data)

    def get_all(self) -> List[Optional[RegionOutput]]:
        return self.repository.get_all()

    def delete(self, _id: UUID4) -> bool:
        if not self.repository.region_exists_by_id(_id):
            raise HTTPException(status_code=404, detail="Region not found")
        region = self.repository.get_by_id(_id)
        self.repository.delete(region)
        return True

    def update(self, _id: UUID4, data: RegionInput) -> RegionInput:
        if not self.repository.region_exists_by_id(_id):
            raise HTTPException(status_code=404, detail="Region not found")
        region = self.repository.get_by_id(_id)
        return self.repository.update(region, data)

service/city_service.py

from typing import List
from fastapi import HTTPException
from pydantic import UUID4
from sqlalchemy.orm import Session
from repository.city_repository import CityRepository
from repository.region_repository import RegionRepository
from schemas.city_schema import CityInput, CityOutput


class CityService:
    """
    Service class for handling cities.
    """
    def __init__(self, session: Session):
        self.repository = CityRepository(session)
        self.region_repository = RegionRepository(session)

    def create(self, data: CityInput) -> CityOutput:
        if self.repository.city_exists_by_name(data.name):
            raise HTTPException(status_code=400, detail="City already exists")
        
        if not self.region_repository.region_exists_by_id(data.region_id):
            raise HTTPException(status_code=400, detail="Region not found")
        
        region = self.region_repository.get_region(data.region_id)
        city = self.repository.create(data)
        return CityOutput(**city.model_dump(exclude_none=True), region=region)

    def get_all_by_region(self, region_id: UUID4) -> List[CityOutput]:
        return self.repository.get_all_by_region(region_id)

    def delete(self, _id: UUID4) -> bool:
        if not self.repository.city_exists_by_id(_id):
            raise HTTPException(status_code=404, detail="City not found")
        city = self.repository.get_by_id(_id)
        return self.repository.delete(city)

    def get_all(self) -> List[CityOutput]:
        return self.repository.get_all()

    def update(self, _id: UUID4, data: CityInput):
        if not self.repository.city_exists_by_id(_id):
            raise HTTPException(status_code=404, detail="City not found")
        city = self.repository.get_by_id(_id)
        updated_city = self.repository.update(city, data)
        return updated_city

Routers

Finally, we define API endpoints using FastAPI routers. These endpoints consume services provided by the service layer and expose them to the client

router/v1/region.py

from typing import List

from fastapi import APIRouter, Depends
from pydantic import UUID4
from sqlalchemy.orm import Session
from config.database import get_db
from schemas.region_schema import RegionOutput, RegionInput
from service.region_service import RegionService


router = APIRouter(
    prefix="/location/region",
    tags=["location"]
)


@router.post("", status_code=201, response_model=RegionOutput)
def create_region(
        data: RegionInput,
        session: Session = Depends(get_db),
):
    _service = RegionService(session)
    return _service.create(data)


@router.get("", status_code=200, response_model=List[RegionOutput])
def get_regions(session: Session = Depends(get_db)) -> List[RegionOutput]:
    _service = RegionService(session)
    return _service.get_all()


@router.delete("/{_id}", status_code=204)
def delete_region(
        _id: UUID4,
        session: Session = Depends(get_db),
):
    _service = RegionService(session)
    return _service.delete(_id)


@router.put("/{_id}", status_code=200, response_model=RegionInput)
def update_region(
        _id: UUID4,
        data: RegionInput,
        session: Session = Depends(get_db),
):
    _service = RegionService(session)
    return _service.update(_id, data)

router/v1/city.py

from typing import List

from fastapi import APIRouter, Depends, HTTPException
from pydantic import UUID4
from sqlalchemy.orm import Session
from config.database import get_db
from schemas.city_schema import CityInput, CityOutput
from service.city_service import CityService


router = APIRouter(
    prefix="/location/city",
    tags=["location"]
)


@router.post("", status_code=201, response_model=CityOutput)
def create_city(
        data: CityInput, session: Session = Depends(get_db),
):
    _service = CityService(session)
    return _service.create(data)


@router.get("/region/{region_id}", status_code=200, response_model=List[CityOutput])
def get_cities_by_region(region_id: UUID4, session: Session = Depends(get_db)):
    _service = CityService(session)
    return _service.get_all_by_region(region_id)


@router.get("", status_code=200, response_model=List[CityOutput])
def get_cities(session: Session = Depends(get_db)):
    _service = CityService(session)
    return _service.get_all()


@router.delete("/{_id}", status_code=204)
def delete_city(
        _id: UUID4,
        session: Session = Depends(get_db),
):
    _service = CityService(session)
    _service.delete(_id):
        

@router.put("/{_id}", status_code=200, response_model=CityInput)
def update_city(
        _id: UUID4,
        data: CityInput,
        session: Session = Depends(get_db),
):
    _service = CityService(session)
    return _service.update(_id, data)

router/api.py

from fastapi import APIRouter
from routers.v1 import region, city

router = APIRouter(
    prefix="/api/v1"
)

router.include_router(region.router)
router.include_router(city.router)