Building a Scalable Project: Service & Repository Patterns

From Django Apps to Layered Architecture

In Django, it's common to group everything related to a feature (models, views, serializers, URLs) into a single "app." This is convenient but can sometimes lead to business logic being scattered across views, model methods, and serializers.

In the FastAPI world, a more common approach is to structure the project by architectural layers, a concept borrowed from Domain-Driven Design (DDD). This promotes a strong separation of concerns.

We'll focus on three core layers:

1. Presentation Layer (Routers): This is the outermost layer. Its only job is to handle HTTP requests and responses. It parses incoming data, calls the business logic layer, and formats the response. It should be "thin," containing no business logic itself.
2. Service Layer (Business Logic): This is the heart of your application. It contains all the business rules and orchestrates the application's behavior. For example: "to create a user, first check if an email is already registered, then hash the password, then save the new user to the database."
3. Data Access Layer (Repositories): This is the innermost layer. Its sole responsibility is to communicate with the data source (in our case, the database). All SELECT, INSERT, UPDATE, and DELETE queries happen here.

This separation makes your code more maintainable, easier to test (you can test business logic without a real database), and more adaptable to change.

To understand the problems this layered approach solves, let's watch a short segment that explains the pitfalls of mixing these concerns.

Building the Project Skeleton

Let's now create the folder structure for our Micro-Blog API. This structure is a widely accepted standard for building scalable FastAPI applications.

In your project's root directory, create the following folders and files:

micro_blog_api/
├── app/
│   ├── __init__.py
│   ├── core/
│   │   └── __init__.py
│   ├── models/
│   │   └── __init__.py
│   ├── schemas/
│   │   └── __init__.py
│   ├── repositories/
│   │   └── __init__.py
│   ├── services/
│   │   └── __init__.py
│   ├── routers/
│   │   └── __init__.py
│   ├── dependencies.py
│   └── main.py
├── tests/
│   └── __init__.py
├── .env
├── alembic.ini
└── requirements.txt

Here's the role of each directory inside app/:

• core/: For application-wide concerns like configuration management (settings.py) and logging setup.
• models/: For your SQLAlchemy database models (the User, Post, and Comment classes we designed).
• schemas/: For your Pydantic schemas (e.g., UserCreate, PostResponse).
• repositories/: The Data Access Layer. Will contain classes that interact directly with the database.
• services/: The Business Logic Layer. Will contain classes that orchestrate calls to repositories.
• routers/: The Presentation Layer. Contains your API endpoints defined with APIRouter.
• dependencies.py: A central place to define reusable dependency injection functions.
• main.py: The entry point of your application, where you create the main FastAPI app instance and include the routers.

The video below walks through a very similar project structure and explains the philosophy behind it.

Implementing the Layers

Now let's see how the repository and service patterns are implemented in code. We'll use the "User" resource as our example.

1. The Repository Pattern

The repository is a class that abstracts away the database queries. It receives a database session and exposes simple methods like create or get_by_email.

Based on that guide, let's sketch out what our UserRepository would look like. You would create a file app/repositories/user_repository.py and write the following code.

app/repositories/user_repository.py

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select

from app.models.user import User  # We'll create this model later
from app.schemas.user import UserCreate # We designed this schema in the last lesson

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def create_user(self, user_data: UserCreate) -> User:
        """
        Creates a new user in the database.
        """
        new_user = User(
            username=user_data.username,
            email=user_data.email,
            hashed_password=user_data.password # Note: hashing will happen in the service!
        )
        self.db.add(new_user)
        await self.db.commit()
        await self.db.refresh(new_user)
        return new_user

    async def get_user_by_email(self, email: str) -> User | None:
        """
        Retrieves a user by their email address.
        """
        result = await self.db.execute(select(User).filter(User.email == email))
        return result.scalars().first()

    async def get_user_by_username(self, username: str) -> User | None:
        """
        Retrieves a user by their username.
        """
        result = await self.db.execute(select(User).filter(User.username == username))
        return result.scalars().first()

Notice the clear, single responsibility: it only knows how to talk to the User table. It doesn't know about password hashing or HTTP errors.

2. The Service Pattern

The service layer coordinates the business logic. It uses one or more repositories to get its job done. It's the "brain" of the operation.

app/services/user_service.py

from fastapi import HTTPException, status
from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate
from app.models.user import User



# We'll add a utility for password hashing later
from app.utils.security import get_password_hash 

class UserService:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo

    async def create_new_user(self, user_data: UserCreate) -> User:
        """
        Orchestrates the user creation process.
        1. Checks for existing user.
        2. Hashes the password.
        3. Calls the repository to create the user.
        """



        # Business Rule: Check if username or email is already taken
        db_user_by_email = await self.user_repo.get_user_by_email(email=user_data.email)
        if db_user_by_email:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Email already registered",
            )
        
        db_user_by_username = await self.user_repo.get_user_by_username(username=user_data.username)
        if db_user_by_username:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Username already taken",
            )




        # Business Rule: Hash the password before saving
        hashed_password = get_password_hash(user_data.password)
        



        # Create a new Pydantic model with the hashed password to pass to the repo
        user_to_create = user_data.copy(update={"password": hashed_password})




        # Call the repository to perform the database operation
        return await self.user_repo.create_user(user_data=user_to_create)

This UserService encapsulates the entire business process for creating a user. The API router will only need to call this one method.

Wiring It All Together with Dependency Injection

So, how does the UserService get an instance of the UserRepository? And how does the UserRepository get a database session? This is where FastAPI's powerful Dependency Injection (DI) system comes in.

We will define functions in app/dependencies.py that FastAPI will use to provide these objects to our path operations.

We won't fully implement the routers and dependencies today, but understanding this concept is crucial. In the next lesson, when we build the user registration endpoint, you will see this in action:

# In app/routers/users.py (for next lesson)
from fastapi import APIRouter, Depends
from app.services.user_service import UserService
from app.dependencies import get_user_service

router = APIRouter()

@router.post("/register")
async def register_user(
    user_data: UserCreate, 
    user_service: UserService = Depends(get_user_service) # DI in action!
):
    new_user = await user_service.create_new_user(user_data)
    return new_user

The router is beautifully thin. It doesn't know how a user is created; it just knows it needs to call the UserService.

Conclusion

You've successfully set up a professional, scalable project structure for your FastAPI application. By separating concerns into distinct layers, you've laid the foundation for a codebase that is clean, easy to test, and a pleasure to maintain.

Key Takeaways:

• Modern FastAPI projects are often structured by architectural layers (Presentation, Service, Data Access) rather than by feature-based apps like in Django.
• The Repository Pattern encapsulates all data access logic, isolating your application from the specifics of the database.
• The Service Layer contains the core business logic, orchestrating calls to repositories and enforcing business rules.
• This layered architecture results in a "thin" router layer, which only handles HTTP-related concerns.
• FastAPI's Dependency Injection system is the glue that seamlessly connects these decoupled layers.

Preview of the Next Lesson:

With our project structure in place, we're ready to implement our first end-to-end feature. In the next lesson, we will focus on user registration and authentication. We'll implement the User SQLAlchemy model, create the dependency functions for the database and services, build the registration endpoint, and implement JWT-based login to issue access tokens.