Hello,
I decided to start sharing what I do with everyone so. Let us 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
Apologies for missing the file structure! Below is the full project structure to help you organize everything:
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
│
├── requirements.txt # Project dependencies
└── README.md # Project description (optional)
File Details
app/main.py: This file contains the FastAPI application setup, where we define the endpoints and handle requests.
app/models.py: This file contains the database models (such as theUsermodel).
app/repository.py: This file contains theBaseRepositoryclass with generic methods for handling CRUD operations.
app/test_repository.py: This file contains the unit tests to verify the functionality of your repository and FastAPI endpoints.
requirements.txt: This file contains the Python dependencies required for the project (FastAPI, SQLAlchemy, etc.).
We will now set up the project and install the necessary dependencies using pip and a requirements.txt file.
1.1 Setting up the requirements.txt
First, create a requirements.txt file in the root of your project directory with the following content:
fastapi==0.95.0
sqlalchemy==2.0.0
sqlmodel==0.0.12
uvicorn==0.18.2
pytest==7.0.0
pytest-asyncio==0.18.3
This file contains the dependencies for our FastAPI app, SQLAlchemy, SQLModel, and testing libraries.
1.2 Installing the Dependencies
Now, install the dependencies by running:
pip install -r requirements.txt
This will allow you to use the project as a local package.
Step 2: Creating the Application Code
We will now build the basic components of our application: the database models, the repository, and the FastAPI app.
2.1 Creating the Database Models
Create a file models.py to define the SQLAlchemy models for the database. For simplicity, we’ll create a User model.
from sqlmodel import SQLModel, Field
import uuid
class User(SQLModel, table=True):
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
username: str
email: str
2.2 Creating the Repository
The repository will handle all database operations, like creating, reading, updating, and deleting records. We will create a BaseRepository class to generalize these operations for any model.
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.3 Creating the FastAPI App
Now we will create the FastAPI app that uses the BaseRepository to interact with the database. Create a file main.py and add the following code:
from fastapi import FastAPI, Depends
from sqlmodel import Session, create_engine, SQLModel
from repository import BaseRepository
from models import User
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: dict, db: Session = Depends(get_db)):
user_repo = BaseRepository(User, db)
return user_repo.create(user_data)
@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: dict, db: Session = Depends(get_db)):
user_repo = BaseRepository(User, db)
return user_repo.update(user_id, user_data)
@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.4 Running the Application
Now that we have our models, repository, and FastAPI app ready, let’s run the application. Use uvicorn to start the server:
uvicorn main:app --reload
This will start the FastAPI app, and you can access the API documentation at http://127.0.0.1:8000/docs.
Step 3: Writing Unit Tests
We will write some basic unit tests to verify that our repository and FastAPI endpoints work as expected.
3.1 Writing the Unit Tests
Create a file test_repository.py and add the following code for the unit tests:
import pytest
from fastapi.testclient import TestClient
from main import app, get_db
from sqlmodel import SQLModel, Session, create_engine, select
from models import User
import uuid
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
def override_get_db():
with Session(engine) as session:
yield session
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
@pytest.fixture(scope="module")
def setup_db():
SQLModel.metadata.create_all(bind=engine)
yield
SQLModel.metadata.drop_all(bind=engine)
def test_create_user(setup_db):
response = client.post("/users/", json={"username": "test_user", "email": "[email protected]"})
assert response.status_code == 200
assert response.json()["username"] == "test_user"
def test_get_user(setup_db):
user_id = uuid.UUID("your-valid-uuid-here")
response = client.get(f"/users/{user_id}")
assert response.status_code == 200
def test_update_user(setup_db):
user_id = uuid.UUID("your-valid-uuid-here")
response = client.put(f"/users/{user_id}", json={"email": "[email protected]"})
assert response.status_code == 200
assert response.json()["email"] == "[email protected]"
def test_delete_user(setup_db):
user_id = uuid.UUID("your-valid-uuid-here")
response = client.delete(f"/users/{user_id}")
assert response.status_code == 200
assert response.json() is True
3.2 Running the Tests
To run the tests, use the following command:
pytest
Conclusion
In this tutorial, we built a simple FastAPI application using the Generic Repository Pattern. We also created a test suite to ensure everything works as expected. 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.
SOCIAL SHARE CARD GENERATOR