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)