Module 6: Project — Advanced Task Manager API

Architecture: Models, Data, and Dependencies

Next →

Overview

This capsule lays the foundations of the Task Manager API: the Pydantic models that define the shape of your data, the in-memory storage with its helpers, and the dependencies that get injected into every endpoint. Without this base, the code in the capsules that follow has nothing to stand on.

Models: the data contracts

Create app/models.py:

from enum import Enum
from datetime import datetime
from pydantic import BaseModel, Field


class Status(str, Enum):
    pending = "pending"
    in_progress = "in_progress"
    completed = "completed"
    cancelled = "cancelled"


class Priority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
    critical = "critical"


class AttachmentInfo(BaseModel):
    filename: str
    original_name: str
    description: str = ""
    content_type: str
    size_bytes: int


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(default="", max_length=1000)
    status: Status = Status.pending
    priority: Priority = Priority.medium
    assignee: str | None = None
    tags: list[str] = Field(default_factory=list)

    model_config = {
        "json_schema_extra": {
            "examples": [{
                "title": "Implement authentication",
                "description": "Add JWT to every protected endpoint",
                "priority": "high",
                "assignee": "mike",
                "tags": ["backend", "security"]
            }]
        }
    }


class TaskUpdate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(default="", max_length=1000)
    status: Status
    priority: Priority
    assignee: str | None = None
    tags: list[str] = Field(default_factory=list)


class TaskPatch(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=1000)
    status: Status | None = None
    priority: Priority | None = None
    assignee: str | None = None
    tags: list[str] | None = None


class TaskResponse(BaseModel):
    id: int
    title: str
    description: str
    status: Status
    priority: Priority
    assignee: str | None
    tags: list[str]
    attachments: list[AttachmentInfo]
    created_at: str
    updated_at: str


class TaskSummary(BaseModel):
    id: int
    title: str
    status: Status
    priority: Priority
    attachments_count: int


class TaskStats(BaseModel):
    total: int
    by_status: dict[str, int]
    by_priority: dict[str, int]
    completion_rate: float
    with_attachments: int

Design decisions

Status and Priority as (str, Enum) — Lets them serialize as strings in JSON and get validated automatically.

Separate models — TaskCreate (input without id/timestamps), TaskUpdate (full input), TaskPatch (everything optional), TaskResponse (output with everything), TaskSummary (condensed output for lists).

AttachmentInfo — A nested model for attached-file metadata.

TaskStats — The model for the statistics endpoint.

Data: in-memory storage

Create app/data.py:

from datetime import datetime


tasks_db: list[dict] = []

audit_log: list[dict] = []


def get_current_timestamp() -> str:
    return datetime.now().isoformat()


def generate_id() -> int:
    if not tasks_db:
        return 1
    return max(t["id"] for t in tasks_db) + 1


def find_task_by_id(task_id: int) -> dict | None:
    return next((t for t in tasks_db if t["id"] == task_id), None)


def seed_data():
    """Loads the initial data for testing."""
    initial_tasks = [
        {
            "id": 1,
            "title": "Set up the FastAPI project",
            "description": "Initial setup with a professional folder structure",
            "status": "completed",
            "priority": "high",
            "assignee": "mike",
            "tags": ["setup", "backend"],
            "attachments": [],
            "created_at": "2026-03-01T10:00:00",
            "updated_at": "2026-03-01T14:30:00"
        },
        {
            "id": 2,
            "title": "Implement the task CRUD",
            "description": "CREATE, READ, UPDATE, DELETE endpoints with validation",
            "status": "in_progress",
            "priority": "high",
            "assignee": "mike",
            "tags": ["backend", "api"],
            "attachments": [],
            "created_at": "2026-03-02T09:00:00",
            "updated_at": "2026-03-02T09:00:00"
        },
        {
            "id": 3,
            "title": "Add WebSocket notifications",
            "description": "Real-time updates when tasks are created or completed",
            "status": "pending",
            "priority": "medium",
            "assignee": None,
            "tags": ["backend", "realtime"],
            "attachments": [],
            "created_at": "2026-03-03T08:00:00",
            "updated_at": "2026-03-03T08:00:00"
        },
        {
            "id": 4,
            "title": "Write the API documentation",
            "description": "A complete README with usage examples and setup",
            "status": "pending",
            "priority": "low",
            "assignee": "alex",
            "tags": ["docs"],
            "attachments": [],
            "created_at": "2026-03-04T11:00:00",
            "updated_at": "2026-03-04T11:00:00"
        },
        {
            "id": 5,
            "title": "Bug: the status filter doesn't work",
            "description": "GET /tasks?status=completed returns every task",
            "status": "in_progress",
            "priority": "critical",
            "assignee": "mike",
            "tags": ["bug", "urgent"],
            "attachments": [],
            "created_at": "2026-03-05T16:00:00",
            "updated_at": "2026-03-05T16:45:00"
        }
    ]

    tasks_db.clear()
    tasks_db.extend(initial_tasks)


def log_audit(action: str, details: dict):
    audit_log.append({
        "action": action,
        "details": details,
        "timestamp": get_current_timestamp()
    })

Dependencies: reusable logic

Create app/dependencies.py:

from fastapi import HTTPException, Query
from app.data import find_task_by_id


def pagination_params(
    skip: int = Query(default=0, ge=0, description="Records to skip"),
    limit: int = Query(default=20, ge=1, le=100, description="Records per page")
) -> dict:
    return {"skip": skip, "limit": limit}


def get_task_or_404(task_id: int) -> dict:
    task = find_task_by_id(task_id)
    if not task:
        raise HTTPException(
            status_code=404,
            detail={
                "error": "Task not found",
                "task_id": task_id,
                "suggestion": "Check the ID with GET /tasks/"
            }
        )
    return task


def filter_params(
    status: str | None = Query(default=None, description="Filter by status"),
    priority: str | None = Query(default=None, description="Filter by priority"),
    assignee: str | None = Query(default=None, description="Filter by assignee"),
    search: str | None = Query(default=None, description="Search in the title and description"),
    tag: str | None = Query(default=None, description="Filter by tag")
) -> dict:
    return {
        "status": status,
        "priority": priority,
        "assignee": assignee,
        "search": search,
        "tag": tag
    }


def apply_filters(tasks: list[dict], filters: dict) -> list[dict]:
    result = tasks

    if filters["status"]:
        result = [t for t in result if t["status"] == filters["status"]]

    if filters["priority"]:
        result = [t for t in result if t["priority"] == filters["priority"]]

    if filters["assignee"]:
        result = [t for t in result if t.get("assignee") == filters["assignee"]]

    if filters["search"]:
        query = filters["search"].lower()
        result = [
            t for t in result
            if query in t["title"].lower() or query in t.get("description", "").lower()
        ]

    if filters["tag"]:
        result = [t for t in result if filters["tag"] in t.get("tags", [])]

    return result

WebSocket Manager

Create app/websocket_manager.py:

from fastapi import WebSocket
from datetime import datetime


class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        if websocket in self.active_connections:
            self.active_connections.remove(websocket)

    async def send_personal(self, data: dict, websocket: WebSocket):
        await websocket.send_json(data)

    async def broadcast(self, data: dict):
        disconnected = []
        for connection in self.active_connections:
            try:
                await connection.send_json(data)
            except Exception:
                disconnected.append(connection)
        for conn in disconnected:
            self.disconnect(conn)

    async def broadcast_event(self, event_type: str, payload: dict):
        await self.broadcast({
            "type": event_type,
            "data": payload,
            "timestamp": datetime.now().isoformat(),
            "connections": len(self.active_connections)
        })

    @property
    def connection_count(self) -> int:
        return len(self.active_connections)


ws_manager = ConnectionManager()

Background notifications

Create app/background/__init__.py (empty) and app/background/notifications.py:

import logging
from datetime import datetime
from app.data import audit_log

logger = logging.getLogger("task_manager.background")


def send_email_notification(
    event: str,
    task_title: str,
    task_id: int,
    recipients: list[str] | None = None
):
    if recipients is None:
        recipients = ["admin@taskmanager.dev"]

    logger.info(
        f"[EMAIL] {event} | Task: '{task_title}' (ID: {task_id}) | "
        f"To: {', '.join(recipients)}"
    )

    audit_log.append({
        "action": "email_sent",
        "details": {
            "event": event,
            "task_id": task_id,
            "task_title": task_title,
            "recipients": recipients
        },
        "timestamp": datetime.now().isoformat()
    })


def log_audit_event(action: str, details: dict):
    logger.info(f"[AUDIT] {action} | {details}")

    audit_log.append({
        "action": action,
        "details": details,
        "timestamp": datetime.now().isoformat()
    })

Middleware

Create app/middleware/__init__.py (empty) and app/middleware/logging.py:

import time
import logging
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

logger = logging.getLogger("task_manager.middleware")


class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()

        response = await call_next(request)

        duration_ms = (time.time() - start_time) * 1000

        logger.info(
            f"{request.method} {request.url.path} "
            f"→ {response.status_code} "
            f"({duration_ms:.1f}ms)"
        )

        response.headers["X-Process-Time-Ms"] = f"{duration_ms:.1f}"

        return response

Routers: init.py

Create app/routers/__init__.py:

from app.routers.tasks import router as tasks_router
from app.routers.uploads import router as uploads_router
from app.routers.websocket import router as ws_router

__all__ = ["tasks_router", "uploads_router", "ws_router"]

Main: assembling everything

Create app/main.py:

import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import tasks_router, uploads_router, ws_router
from app.middleware.logging import LoggingMiddleware
from app.data import seed_data

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s"
)


@asynccontextmanager
async def lifespan(app):
    seed_data()
    logging.getLogger("task_manager").info("Task Manager API started with sample data")
    yield
    logging.getLogger("task_manager").info("Task Manager API stopped")


app = FastAPI(
    title="Task Manager API",
    description="""
## An advanced API for task management

Features:
- **Full CRUD** with filters, pagination, and search
- **WebSocket** for real-time notifications
- **File uploads** to attach documents to tasks
- **Background tasks** for email notifications
- **CSV export** of tasks

### WebSocket
Connect to `ws://host/ws` to receive events in real time.

### Authentication
This version doesn't include auth (see guide #9).
    """,
    version="2.0.0",
    lifespan=lifespan
)

app.add_middleware(LoggingMiddleware)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(tasks_router)
app.include_router(uploads_router)
app.include_router(ws_router)


@app.get("/health", tags=["System"])
async def health_check():
    from app.websocket_manager import ws_manager
    from app.data import tasks_db, audit_log

    return {
        "status": "healthy",
        "version": "2.0.0",
        "tasks_count": len(tasks_db),
        "websocket_connections": ws_manager.connection_count,
        "audit_events": len(audit_log)
    }

Verifying step 1

# Create the structure
mkdir -p app/routers app/middleware app/background app/uploads
touch app/__init__.py app/routers/__init__.py app/middleware/__init__.py app/background/__init__.py

# Check that the imports work
python -c "from app.models import TaskCreate, TaskResponse, Status, Priority; print('Models OK')"
python -c "from app.data import tasks_db, seed_data; seed_data(); print(f'Data OK: {len(tasks_db)} tasks')"
python -c "from app.dependencies import pagination_params, get_task_or_404; print('Dependencies OK')"
python -c "from app.websocket_manager import ws_manager; print('WS Manager OK')"

If everything prints "OK", the base is ready.

What's next?

In Capsule 03 you implement all the CRUD endpoints with advanced response models and background tasks wired in.

← PreviousFinal Project: Advanced Task Manager APINext →CRUD Endpoints, Responses, and Background Tasks