Modularising FastAPI: Controller, Service, Repository

Once your application grows beyond a handful of endpoints, you’ll notice that route functions start doing too much — handling HTTP requests, applying business rules, and talking to the data store all in the same function. This works for small projects, but it makes code hard to test, reuse, and change. The solution is a layered architecture — also known as the Controller–Service–Repository (CSR) pattern — where each layer has exactly one responsibility. This lesson refactors the Employee Management System step by step into this professional structure.

Why Modularise?

Consider what creating an employee actually involves:

Receiving the HTTP request
Validating the incoming data
Applying business rules (generate IDs, employee codes, timestamps)
Saving the employee to the data store
Returning the response

Before modularisation, all of this lives in one function:

@app.post("/employees")
def create_employee(employee: EmployeeCreate):
    new_id = max(EMPLOYEES.keys()) + 1
    EMPLOYEES[new_id] = {"id": new_id, **employee.model_dump()}
    return EMPLOYEES[new_id]

This is fine at first, but as features are added — validation rules, duplicate checks, email notifications, audit logs — this function grows unmanageable. After modularisation, each concern lives in its own layer:

Client
   │
   ▼
Controller (HTTP Layer)
   │
   ▼
Service (Business Logic)
   │
   ▼
Repository (Data Access)
   │
   ▼
Database

Layer	Responsibility
Controller (Router)	Receives HTTP requests, validates input, returns HTTP responses
Service	Implements business logic, generates IDs, applies rules
Repository	Reads and writes data, converts between raw data and models
Database	Stores the application’s data

Project Structure

app/
├── main.py
├── database.py
├── dependencies.py
├── models/
│   ├── __init__.py
│   └── employee.py
├── repositories/
│   ├── __init__.py
│   └── employee_repository.py
├── services/
│   ├── __init__.py
│   └── employee_service.py
└── routers/
    ├── __init__.py
    └── employee_router.py

Step 1: Define the Models

Each layer works with data differently, so you define separate models for incoming requests, internal processing, and outgoing responses.

app/models/employee.py

from datetime import datetime
from pydantic import BaseModel, Field


# -------------------------
# Request Models
# -------------------------

class EmployeeCreateRequest(BaseModel):
    name: str = Field(..., min_length=2)
    department: str
    role: str


class EmployeeUpdateRequest(BaseModel):
    name: str | None = None
    department: str | None = None
    role: str | None = None


# -------------------------
# Business Model
# -------------------------

class EmployeeBusiness(BaseModel):
    id: int
    employee_code: str
    name: str
    department: str
    role: str
    created_at: datetime


# -------------------------
# Response Model
# -------------------------

class EmployeeResponse(BaseModel):
    id: int
    name: str
    department: str
    role: str

Client Request
       │
       ▼
EmployeeCreateRequest
       │
       ▼
EmployeeBusiness
       │
       ▼
EmployeeResponse
       │
       ▼
Client Response

Step 2: Repository Layer

The Repository bridges the Service and the data source. It hides the implementation details of how data is stored — whether in a dictionary, PostgreSQL, or MongoDB. For now it uses an in-memory dictionary. The Repository uses Constructor Injection — the data source is passed in from outside rather than created internally:

app/database.py

EMPLOYEES = {
    1: {
        "id": 1,
        "employee_code": "EMP001",
        "name": "Alice Smith",
        "department": "Engineering",
        "role": "Backend Developer",
        "created_at": "2024-01-15T09:00:00",
    },
    2: {
        "id": 2,
        "employee_code": "EMP002",
        "name": "Bob Jones",
        "department": "Product",
        "role": "Product Manager",
        "created_at": "2024-01-16T10:00:00",
    },
}

app/repositories/employee_repository.py

from app.models.employee import EmployeeBusiness


class EmployeeRepository:

    def __init__(self, employees: dict[int, dict]):
        self.employees = {
            emp_id: EmployeeBusiness(**emp)
            for emp_id, emp in employees.items()
        }

    def get_all(self) -> list[EmployeeBusiness]:
        return list(self.employees.values())

    def get_by_id(self, employee_id: int) -> EmployeeBusiness | None:
        return self.employees.get(employee_id)

    def create(self, employee: EmployeeBusiness) -> EmployeeBusiness:
        self.employees[employee.id] = employee
        return employee

    def update(self, employee: EmployeeBusiness) -> EmployeeBusiness:
        self.employees[employee.id] = employee
        return employee

    def delete(self, employee_id: int) -> None:
        del self.employees[employee_id]

Step 3: FastAPI Dependency Injection

Instead of manually creating the Repository in every endpoint, define a dependency provider function and let FastAPI call it automatically via Depends():

app/dependencies.py

from typing import Annotated
from fastapi import Depends
from app.database import EMPLOYEES
from app.repositories.employee_repository import EmployeeRepository
from app.services.employee_service import EmployeeService


def get_employee_repository() -> EmployeeRepository:
    return EmployeeRepository(EMPLOYEES)


RepositoryDep = Annotated[
    EmployeeRepository,
    Depends(get_employee_repository),
]


def get_employee_service(repository: RepositoryDep) -> EmployeeService:
    return EmployeeService(repository)


ServiceDep = Annotated[
    EmployeeService,
    Depends(get_employee_service),
]

EMPLOYEES (database.py)
      │
      ▼
get_employee_repository()
      │
      ▼
EmployeeRepository
      │
      ▼
get_employee_service()
      │
      ▼
EmployeeService
      │
      ▼
Your Endpoint

Step 4: Service Layer

The Service contains the business logic. It receives the Repository through its constructor and transforms Request Models into Business Models.

app/services/employee_service.py

from datetime import UTC, datetime
from app.models.employee import EmployeeBusiness, EmployeeCreateRequest, EmployeeUpdateRequest
from app.repositories.employee_repository import EmployeeRepository


class EmployeeService:

    def __init__(self, repository: EmployeeRepository):
        self.repository = repository

    def get_all(self) -> list[EmployeeBusiness]:
        return self.repository.get_all()

    def get_by_id(self, employee_id: int) -> EmployeeBusiness | None:
        return self.repository.get_by_id(employee_id)

    def create(self, request: EmployeeCreateRequest) -> EmployeeBusiness:
        employees = self.repository.get_all()
        next_id = max((e.id for e in employees), default=0) + 1

        employee = EmployeeBusiness(
            id=next_id,
            employee_code=f"EMP{next_id:03}",
            name=request.name,
            department=request.department,
            role=request.role,
            created_at=datetime.now(UTC),
        )
        return self.repository.create(employee)

    def update(self, employee_id: int, request: EmployeeUpdateRequest) -> EmployeeBusiness | None:
        employee = self.repository.get_by_id(employee_id)
        if employee is None:
            return None

        for field, value in request.model_dump(exclude_unset=True).items():
            setattr(employee, field, value)

        return self.repository.update(employee)

    def delete(self, employee_id: int) -> None:
        self.repository.delete(employee_id)

Step 5: Router (Controller) Layer

The Router is the entry point of every HTTP request. It validates inputs, delegates to the Service, and returns structured responses. It contains no business logic.

app/routers/employee_router.py

from fastapi import APIRouter, HTTPException, status
from app.dependencies import ServiceDep
from app.models.employee import EmployeeCreateRequest, EmployeeResponse, EmployeeUpdateRequest

router = APIRouter(prefix="/employees", tags=["Employees"])


@router.get("/", response_model=list[EmployeeResponse])
def get_employees(service: ServiceDep):
    return service.get_all()


@router.get("/{employee_id}", response_model=EmployeeResponse)
def get_employee(employee_id: int, service: ServiceDep):
    employee = service.get_by_id(employee_id)
    if employee is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Employee not found.")
    return employee


@router.post("/", response_model=EmployeeResponse, status_code=status.HTTP_201_CREATED)
def create_employee(request: EmployeeCreateRequest, service: ServiceDep):
    return service.create(request)


@router.put("/{employee_id}", response_model=EmployeeResponse)
def update_employee(employee_id: int, request: EmployeeUpdateRequest, service: ServiceDep):
    employee = service.update(employee_id, request)
    if employee is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Employee not found.")
    return employee


@router.delete("/{employee_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_employee(employee_id: int, service: ServiceDep):
    if service.get_by_id(employee_id) is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Employee not found.")
    service.delete(employee_id)

app/main.py

from fastapi import FastAPI
from app.routers.employee_router import router as employee_router

app = FastAPI(title="Employee Management System")

app.include_router(employee_router)

Complete Request Flow

Client
   │
   ▼
Employee Router      ← HTTP layer, validates input with EmployeeCreateRequest
   │
   ▼
Employee Service     ← Business logic, creates EmployeeBusiness with generated fields
   │
   ▼
Employee Repository  ← Data access, stores/retrieves EmployeeBusiness objects
   │
   ▼
Database (EMPLOYEES dict)

Benefits of This Architecture

Benefit	Why it matters
Replaceability	Switch from in-memory dict to PostgreSQL by changing only the Repository
Testability	Test the Service with a mock Repository — no HTTP or database needed
Readability	Each file has one job; its purpose is obvious from the file name
Extensibility	Add new business rules in the Service without touching the Router or Repository

This is the architecture used by most production FastAPI applications. Once you connect a real database in the next lesson, you’ll only need to modify the Repository layer — the Router and Service remain exactly the same.

SQL & Databases

Data Analysis

FastAPI

Frontend

Modularising FastAPI: Controller, Service, Repository

Why Modularise?

Project Structure

Step 1: Define the Models

Step 2: Repository Layer

Step 3: FastAPI Dependency Injection

Step 4: Service Layer

Step 5: Router (Controller) Layer

Complete Request Flow

Benefits of This Architecture

​Why Modularise?

​Project Structure

​Step 1: Define the Models

​Step 2: Repository Layer

​Step 3: FastAPI Dependency Injection

​Step 4: Service Layer

​Step 5: Router (Controller) Layer

​Complete Request Flow

​Benefits of This Architecture

Why Modularise?

Project Structure

Step 1: Define the Models

Step 2: Repository Layer

Step 3: FastAPI Dependency Injection

Step 4: Service Layer

Step 5: Router (Controller) Layer

Complete Request Flow

Benefits of This Architecture