Reusable Components and Services¶

🎯 What You’ll Learn¶

Why separating business logic from routing improves maintainability
How to build service functions that encapsulate core logic
How to inject services using FastAPI’s Depends system
How to structure your project for reuse and testing

🧠 Why Use Services?¶

In a real-world API, routes should focus on:

HTTP concerns (status codes, request/response)
Dependency injection
Input/output validation

Business logic — like querying the database, applying rules, or transforming data — should live in service functions.

Benefits:

✅ Easier to test
✅ Easier to reuse across routes
✅ Easier to refactor and extend

🗂️ Suggested Project Structure¶

task_manager/
├── main.py
├── db.py
├── models/
│   └── task.py
├── routers/
│   └── tasks.py
├── services/
│   └── task_service.py

🧱 Step 1: Move Business Logic to a Service¶

📄 services/task_service.py

from sqlmodel import Session, select
from models.task import Task
from fastapi import HTTPException

def get_task_by_id(task_id: int, session: Session) -> Task:
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

def list_tasks(session: Session) -> list[Task]:
    return session.exec(select(Task)).all()

def create_task(task: Task, session: Session) -> Task:
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

def update_task(task_id: int, updated: Task, session: Session) -> Task:
    task = get_task_by_id(task_id, session)
    task.title = updated.title
    task.description = updated.description
    task.completed = updated.completed
    session.commit()
    session.refresh(task)
    return task

def delete_task(task_id: int, session: Session) -> None:
    task = get_task_by_id(task_id, session)
    session.delete(task)
    session.commit()

🔁 Step 2: Use Services in Your Router¶

📄 routers/tasks.py

from fastapi import APIRouter, Depends
from sqlmodel import Session
from db import get_session
from models.task import Task
from services import task_service

router = APIRouter(prefix="/tasks", tags=["Tasks"])

@router.get("/")
def get_all(session: Session = Depends(get_session)):
    return task_service.list_tasks(session)

@router.get("/{task_id}")
def get_one(task_id: int, session: Session = Depends(get_session)):
    return task_service.get_task_by_id(task_id, session)

@router.post("/", status_code=201)
def create(task: Task, session: Session = Depends(get_session)):
    return task_service.create_task(task, session)

@router.put("/{task_id}")
def update(task_id: int, updated: Task, session: Session = Depends(get_session)):
    return task_service.update_task(task_id, updated, session)

@router.delete("/{task_id}", status_code=204)
def delete(task_id: int, session: Session = Depends(get_session)):
    task_service.delete_task(task_id, session)

🧠 Best Practices¶

Keep service functions pure and focused on business logic
Avoid using HTTPException in services unless it’s part of the domain logic
Use Depends to inject shared resources like sessions or authenticated users
Write unit tests for services independently from routes

🧪 Practice Challenge¶

Create a new service:

user_service.py with functions to create and fetch users
Inject it into a users.py router
Use Depends(get_session) to pass the session into service functions