Building Your First API — Task Manager (Part 2)¶

🎯 What You’ll Build¶

A fully persistent task manager backed by SQLite and SQLModel
Filtering by task status (e.g. completed or not)
Pagination for large task lists
Clean separation of routing and business logic using routers and services

🧱 Step 1: Persist Tasks in a Database¶

You’ve already defined your Task model and created the table using SQLModel. Now let’s make sure all CRUD operations use the database.

📄 services/task_service.py

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

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

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 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: Add Filtering by Completion Status¶

Let’s add a query parameter to filter tasks by their completed status.

📄 services/task_service.py

def list_tasks(session: Session, completed: bool | None = None) -> list[Task]:
    query = select(Task)
    if completed is not None:
        query = query.where(Task.completed == completed)
    return session.exec(query).all()

📄 routers/tasks.py

@router.get("/")
def get_all(completed: bool | None = None, session: Session = Depends(get_session)):
    return task_service.list_tasks(session, completed)

🧪 Try:

/tasks → returns all tasks
/tasks?completed=true → returns only completed tasks

📊 Step 3: Add Pagination¶

Let’s add limit and offset parameters to control how many tasks are returned.

📄 services/task_service.py

def list_tasks_paginated(
    session: Session,
    completed: bool | None = None,
    limit: int = 10,
    offset: int = 0
) -> list[Task]:
    query = select(Task)
    if completed is not None:
        query = query.where(Task.completed == completed)
    query = query.offset(offset).limit(limit)
    return session.exec(query).all()

📄 routers/tasks.py

@router.get("/paginated")
def get_paginated(
    completed: bool | None = None,
    limit: int = 10,
    offset: int = 0,
    session: Session = Depends(get_session)
):
    return task_service.list_tasks_paginated(session, completed, limit, offset)

🧪 Try:

/tasks/paginated?limit=5&offset=0
/tasks/paginated?completed=false&limit=3&offset=6

🧩 Step 4: Refactor with Routers and Services¶

Your project structure should now look like this:

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

📄 main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from db import engine
from sqlmodel import SQLModel
from routers import tasks

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- Startup Logic ---
    # This runs before the application starts taking requests
    SQLModel.metadata.create_all(engine)

    yield  # The application runs while stuck here

    # --- Shutdown Logic ---
    # This runs after the application finishes handling requests
    # (e.g., close DB connections, clean up resources)
    pass

app = FastAPI(lifespan=lifespan)

app.include_router(tasks.router)

📚 Swagger UI¶

Visit http://127.0.0.1:8000/docs to:

Explore all endpoints
Test filtering and pagination
See request/response schemas

🧠 Recap¶

You now have a robust, modular API with:

Feature	Endpoint	Description
Create Task	`POST /tasks`	Add a new task
Read All Tasks	`GET /tasks`	List tasks with optional filtering
Paginated Tasks	`GET /tasks/paginated`	List tasks with limit/offset
Read One Task	`GET /tasks/{id}`	Get task by ID
Update Task	`PUT /tasks/{id}`	Replace task
Delete Task	`DELETE /tasks/{id}`	Remove task

All backed by a real database and cleanly separated into routers and services.

🧪 Practice Challenge¶

Extend the API with:

A PATCH /tasks/{id}/toggle endpoint to flip the completed status
A GET /tasks/stats endpoint that returns counts of completed and pending tasks
A GET /tasks/search?query=milk endpoint that filters by title substring