Messanga11

Posted on Jan 23

Generic Repository in FastApi that is managing relationships - PART 1

Hello,

I decided to start sharing what I do with everyone. Let’s get started!

Introduction

In this tutorial, we will build a simple FastAPI application using the Generic Repository Pattern. The repository pattern helps us manage access to data in a centralized and reusable way, making it easy to interact with our database models. We will focus on a simple base code to demonstrate the repository functionality, and we'll also cover unit tests to verify everything works as expected.

What is the Repository Pattern?

The Repository Pattern is a design pattern that abstracts the data access logic in your application. Instead of directly interacting with your database in multiple places, you centralize the logic in a repository class, making it more manageable, reusable, and easier to test.

Step 1: Setting Up the Project

1.1 Create a Virtual Environment

First, let’s create a virtual environment to isolate our project dependencies:

python -m venv venv

Activate the virtual environment:

On macOS/Linux:

  source venv/bin/activate

On Windows:

  venv\Scripts\activate

1.2 Project Structure

Here’s the full project structure:

my_fastapi_project/
│
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI application setup
│   ├── models.py        # Database models
│   ├── repository.py    # Base repository logic
│   └── test_repository.py  # Unit tests for the repository and endpoints
│
├── alembic/             # Alembic migrations folder
│   ├── versions/        # Migration scripts
│   └── env.py           # Alembic environment configuration
│
├── requirements.txt     # Project dependencies
└── README.md            # Project description (optional)

1.3 Setting Up Dependencies

Create a requirements.txt file in the root of your project directory with the following content:

alembic==1.14.1
annotated-types==0.7.0
anyio==4.8.0
click==8.1.8
fastapi==0.115.7
h11==0.14.0
idna==3.10
iniconfig==2.0.0
Mako==1.3.8
MarkupSafe==3.0.2
packaging==24.2
pluggy==1.5.0
pydantic==2.10.5
pydantic_core==2.27.2
pytest==8.3.4
pytest-asyncio==0.25.2
sniffio==1.3.1
SQLAlchemy==2.0.37
sqlmodel==0.0.22
starlette==0.45.2
typing_extensions==4.12.2
uvicorn==0.34.0

Install the dependencies:

pip install -r requirements.txt

Step 2: Creating the Application Code

2.1 Creating the Database Models

Create a file models.py to define the SQLModel models for the database. For simplicity, we’ll create a User model.

from sqlmodel import SQLModel, Field
import uuid
from typing import Optional


class User(SQLModel, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    username: str
    email: str


class UserCreate(SQLModel):
    username: str
    email: str


class UserUpdate(SQLModel):
    username: Optional[str]
    email: Optional[str]

2.2 Setting Up Alembic for Migrations

Initialize Alembic for database migrations:

alembic init alembic

Update the alembic.ini file to point to your database:

sqlalchemy.url = sqlite:///./test.db

Update the alembic/env.py file to use SQLModel:

# Add this import
from app.models import SQLModel

# Search target_metadata and change like this
target_metadata = SQLModel.metadata

You may need to update the script.py.mako file like this

import sqlalchemy as sa
# Add this line below
import sqlmodel

Create your first migration:

alembic revision --autogenerate -m "Initial migration"

Apply the migration:

alembic upgrade head

2.3 Creating the Repository

Create a file repository.py and add the following code:

from typing import Type, TypeVar, Generic, List, Dict, Any
from sqlmodel import Session, select, SQLModel
from fastapi import HTTPException
import uuid

T = TypeVar("T", bound=SQLModel)


class BaseRepository(Generic[T]):
    def __init__(self, model: Type[T], session: Session):
        self.model = model
        self.session = session

    def create(self, data: Dict[str, Any], commit=True) -> T:
        try:
            obj = self.model(**data)
            self.session.add(obj)
            if commit:
                self.session.commit()
                self.session.refresh(obj)
            return obj
        except Exception as e:
            self.session.rollback()
            raise HTTPException(status_code=400, detail=str(e))

    def get(self, id: uuid.UUID) -> T:
        return self.session.get(self.model, id)

    def get_all(self) -> List[T]:
        statement = select(self.model)
        return self.session.exec(statement).all()

    def update(self, id: uuid.UUID, data: Dict[str, Any], commit=True) -> T:
        obj = self.session.get(self.model, id)
        if not obj:
            raise HTTPException(status_code=404, detail="Item not found")

        for key, value in data.items():
            setattr(obj, key, value)

        if commit:
            self.session.commit()
            self.session.refresh(obj)

        return obj

    def delete(self, id: uuid.UUID, commit=True) -> bool:
        obj = self.session.get(self.model, id)
        if not obj:
            raise HTTPException(status_code=404, detail="Item not found")

        self.session.delete(obj)
        if commit:
            self.session.commit()
        return True

2.4 Creating the FastAPI App

Create a file main.py and add the following code:

from fastapi import FastAPI, Depends
from sqlmodel import Session, create_engine
from app.repository import BaseRepository
from app.models import User, UserCreate, UserUpdate
import uuid

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)


def get_db():
    with Session(engine) as session:
        yield session


app = FastAPI()


@app.post("/users")
def create_user(user_data: UserCreate, db: Session = Depends(get_db)):
    user_repo = BaseRepository(User, db)
    return user_repo.create(user_data.model_dump())


@app.get("/users/{user_id}")
def get_user(user_id: uuid.UUID, db: Session = Depends(get_db)):
    user_repo = BaseRepository(User, db)
    return user_repo.get(user_id)


@app.get("/users")
def get_all_users(db: Session = Depends(get_db)):
    user_repo = BaseRepository(User, db)
    return user_repo.get_all()


@app.put("/users/{user_id}")
def update_user(
    user_id: uuid.UUID,
    user_data: UserUpdate,
    db: Session = Depends(get_db),
):
    user_repo = BaseRepository(User, db)
    return user_repo.update(user_id, user_data.model_dump())


@app.delete("/users/{user_id}")
def delete_user(user_id: uuid.UUID, db: Session = Depends(get_db)):
    user_repo = BaseRepository(User, db)
    return user_repo.delete(user_id)

2.5 Running the Application

Start the FastAPI app using uvicorn:

uvicorn app.main:app --reload

You can access the API documentation at http://127.0.0.1:8000/docs.

Conclusion

In this tutorial, we built a simple FastAPI application using the Generic Repository Pattern. We also added Alembic for database migrations. The repository pattern allows us to manage data access in a centralized and reusable manner, making our code cleaner and easier to maintain.

You can extend this application by adding more models, services, and more complex logic as needed. The concepts demonstrated here will form the foundation for a scalable and organized FastAPI project.

DEV Community