Building a Service Layer

The Three-Layer Architecture

In the last lesson, our code had two primary concerns: HTTP handling (in the router) and database access (in the repository). We're now introducing a third layer in the middle to handle business logic.

Let's break down the responsibilities:

1. Presentation Layer (Router): Responsible for HTTP-related tasks. It parses request bodies, extracts path parameters, handles authentication headers, and formats the final JSON response. It should contain no business logic.
2. Service Layer (Business Logic): This is the core of your application. It executes business rules, performs validations, and orchestrates operations. For example, creating a user isn't just an INSERT statement; it involves checking for duplicates, hashing a password, and perhaps sending a welcome email. This all belongs in the service layer.
3. Repository/Data Access Layer: Responsible only for communicating with the database. Its methods should be simple and reflect data storage operations like create, get_by_id, update, etc.

This separation of concerns is the main motivation for this architecture. It makes the application easier to understand, maintain, and test.

Let's begin by reading a short article that clearly defines these layers and their purpose within a FastAPI context.

Implementing the Service Layer

Let's continue with the create_user example. After the last lesson, our route handler might look like this:

# app/routers/users.py - BEFORE the service layer

@router.post("/users", response_model=UserRead)
async def create_user(
    user: UserCreate, 
    repo: UserRepository = Depends(get_user_repository)
):



    # This is BUSINESS LOGIC
    db_user = await repo.get_user_by_email(email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered.")
    



    # This is also BUSINESS LOGIC
    hashed_password = get_password_hash(user.password)
    



    # This is a call to the data layer
    return await repo.create_user(user_data=user, hashed_password=hashed_password)

Our goal is to move all the business logic into a new UserService.

Step 1: Create the UserService

First, let's watch a brief video that visualizes where the service layer fits and what it contains.

Inspired by that, we'll create app/services/user_service.py. The UserService class will depend on the UserRepository we built previously. This is a key concept: services depend on repositories, not the other way around.

# app/services/user_service.py

from fastapi import HTTPException
from app.repositories.user_repository import UserRepository
from app.schemas.user import UserCreate  # Pydantic schema for request
from app.core.security import get_password_hash  # A helper function for hashing

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



        # The service is initialized with a repository instance
        self.user_repo = user_repo

    async def create_user(self, user_data: UserCreate):
        """
        Orchestrates the user creation process.
        """



        # Business Rule 1: Check if a user with this email already exists
        existing_user = await self.user_repo.get_user_by_email(user_data.email)
        if existing_user:
            raise HTTPException(
                status_code=400, 
                detail="An account with this email already exists."
            )
        



        # Business Rule 2: Hash the incoming password
        hashed_password = get_password_hash(user_data.password)




        # Orchestration: Call the repository to save the new user
        # Note: we pass the hashed password, not the original user_data object
        new_user = await self.user_repo.create_user(
            email=user_data.email, 
            hashed_password=hashed_password
        )
        
        return new_user

Notice how this create_user method clearly outlines the business process, free from any HTTP or direct database code.

Step 2: Wire up the Service with Dependency Injection

Now that we have our UserService, how does our router get access to it? Just as with the repository, we'll use FastAPI's dependency injection system. We'll create a new dependency function that builds and provides a UserService instance.

In app/dependencies.py, add the following:

# app/dependencies.py

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.repositories.user_repository import UserRepository
from app.services.user_service import UserService




# Dependency from previous lesson
def get_user_repository(db: AsyncSession = Depends(get_db)) -> UserRepository:
    return UserRepository(db)




# New dependency for the service layer
def get_user_service(
    repo: UserRepository = Depends(get_user_repository)
) -> UserService:
    return UserService(repo)

This new get_user_service function is beautiful in its simplicity. It declares its own dependency on get_user_repository. FastAPI will handle this chain automatically:

1. A request comes in for create_user.
2. FastAPI sees it Depends on get_user_service.
3. To resolve get_user_service, FastAPI sees it Depends on get_user_repository.
4. To resolve get_user_repository, FastAPI sees it Depends on get_db.
5. get_db yields a session, which is passed to get_user_repository to create a UserRepository.
6. The UserRepository instance is passed to get_user_service to create a UserService.
7. The UserService instance is finally passed to our route handler.

Step 3: Refactor the Route Handler

With the service and dependency in place, our create_user route handler becomes remarkably clean and focused.

# app/routers/users.py - AFTER implementing the service layer

from app.services.user_service import UserService
from app.dependencies import get_user_service

@router.post("/users", response_model=UserRead)
async def create_user(
    user: UserCreate, 
    service: UserService = Depends(get_user_service) # Depend on the service now
):



    # The router's only job: delegate to the service and return the result.
    return await service.create_user(user_data=user)

Look at that! The router now has a single responsibility: handling the HTTP request and delegating the real work to the service layer. It has no knowledge of the underlying business rules or data storage mechanisms.

The Payoff: Clarity, Maintainability, and Testability

You've now seen the complete flow from a "fat" endpoint to a clean, three-layer architecture. Let's read one more resource that reinforces why this is so valuable and provides another concrete example.

The benefits described in the article are worth restating:

• Separation of Concerns: Business logic is centralized, not scattered across endpoints.
• Decoupling: The presentation layer (routers) is decoupled from the business logic. You could add a CLI or a gRPC interface that reuses the exact same UserService!
• Testability: This is a huge win. Testing the UserService is now trivial. You can pass it a mock InMemoryUserRepository (like we created in the last lesson) and test your business rules in complete isolation from any database or HTTP server.

# tests/test_user_service.py




# A mock repository for testing
class MockUserRepository:
    async def get_user_by_email(self, email: str):
        if email == "exists@example.com":
            return User(id=1, email=email) # Return a dummy user
        return None

    async def create_user(self, email: str, hashed_password: str):
        return User(id=2, email=email) # Return another dummy user

async def test_create_user_fails_if_email_exists():
    mock_repo = MockUserRepository()
    service = UserService(user_repo=mock_repo)
    user_data = UserCreate(email="exists@example.com", password="password")




    # Use pytest.raises to assert that an HTTPException is raised
    with pytest.raises(HTTPException) as excinfo:
        await service.create_user(user_data)
    
    assert excinfo.value.status_code == 400
    assert "already exists" in excinfo.value.detail

These kinds of unit tests are incredibly fast and precise, allowing you to validate your core business logic with confidence.

Conclusion

In this lesson, you have successfully implemented the service layer, completing the trifecta of a modern, layered API architecture: Router, Service, and Repository. You now have a clear mental model and a practical implementation for separating concerns within your FastAPI applications.

Key Takeaways:

• The Service Layer is responsible for encapsulating all business logic and orchestrating application workflows.
• Services depend on repositories to perform data persistence, but they are not concerned with the details of database interaction.
• Routers depend on services to execute business logic. This makes routers extremely thin, focused only on HTTP-related tasks.
• This layered architecture is wired together elegantly using FastAPI's dependency injection system, with dependencies chaining from the router down to the database session.
• The primary benefit is separation of concerns, which leads to a more maintainable, scalable, and, most importantly, testable codebase.

Preview of the Next Lesson:

We've now successfully refactored a single CRUD endpoint to use this clean architecture. The next logical step is to see how this pattern scales. In our next lesson, we will refactor a simple CRUD endpoint to use the service-repository pattern, solidifying your understanding and demonstrating how this structure applies to all the operations (Create, Read, Update, Delete) for a given resource.