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

🎯 Learning Goals¶

By the end of this lesson, you will be able to:

Trigger notifications using FastAPI background tasks.
Deliver real-time task updates to connected clients via WebSockets.
Write unit and integration tests to ensure full coverage of these features.

⚡ Step 1: Notifications via Background Tasks¶

Background tasks let you perform work after returning a response — perfect for sending notifications without slowing down the user.

📄 routers/tasks.py

from fastapi import APIRouter, Depends, BackgroundTasks
from sqlmodel import Session, select
from db import get_session
from models.task import Task
from models.user import User
from core.dependencies import get_current_user

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

def send_notification(user: User, message: str):
    # Example: log to file or send email
    with open("notifications.log", "a") as f:
        f.write(f"Notify {user.username}: {message}\n")

@router.post("/", status_code=201)
def create_task(
    task: Task,
    background_tasks: BackgroundTasks,
    session: Session = Depends(get_session),
    user: User = Depends(get_current_user)
):
    task.user_id = user.id
    session.add(task)
    session.commit()
    session.refresh(task)

    # Schedule notification
    background_tasks.add_task(send_notification, user, f"Task created: {task.title}")

    return task

🔍 What’s happening?¶

BackgroundTasks is injected into the endpoint.
send_notification() is scheduled after the response.
The user gets a fast response, while the notification is processed in the background.

⚡ Step 2: Real-Time Updates with WebSockets¶

WebSockets allow persistent, two-way communication. We’ll broadcast task updates to all connected clients.

📄 routers/ws.py

from fastapi import APIRouter, WebSocket
from typing import List

router = APIRouter(prefix="/ws", tags=["WebSockets"])

active_connections: List[WebSocket] = []

async def broadcast(message: str):
    for connection in active_connections:
        await connection.send_text(message)

@router.websocket("/tasks")
async def task_updates(websocket: WebSocket):
    await websocket.accept()
    active_connections.append(websocket)
    try:
        while True:
            await websocket.receive_text()  # keep alive
    except:
        active_connections.remove(websocket)

📄 Update routers/tasks.py to broadcast:

from routers.ws import broadcast

@router.post("/", status_code=201)
async def create_task(...):
    ...
    await broadcast(f"Task created: {task.title}")
    return task

🔍 What’s happening?¶

Clients connect to /ws/tasks.
active_connections tracks connected clients.
broadcast() sends updates to all clients when tasks are created or updated.

⚡ Step 3: Full Test Coverage¶

We’ll use pytest and TestClient to test background tasks and WebSockets.

📄 tests/test_tasks.py

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_task_triggers_notification(tmp_path):
    # Override notification log path
    log_file = tmp_path / "notifications.log"

    def fake_notify(user, message):
        with open(log_file, "a") as f:
            f.write(message)

    app.dependency_overrides[send_notification] = fake_notify

    response = client.post("/tasks/", json={"title": "Test Task"})
    assert response.status_code == 201

    # Check notification was written
    with open(log_file) as f:
        content = f.read()
    assert "Task created: Test Task" in content

📄 tests/test_ws.py

import pytest
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

@pytest.mark.asyncio
async def test_websocket_task_updates():
    with client.websocket_connect("/ws/tasks") as websocket:
        # Simulate task creation
        response = client.post("/tasks/", json={"title": "WS Task"})
        assert response.status_code == 201

        # Receive broadcast
        message = websocket.receive_text()
        assert "Task created: WS Task" in message

🧠 Best Practices¶

Use background tasks for non-critical work (notifications, logging).
Use WebSockets for real-time features (live updates, chat).
Always write tests for new features:
- Unit tests for isolated logic.
- Integration tests for API endpoints.
- WebSocket tests for real-time flows.

🧪 Practice Challenge¶

Extend notifications to send an email (mocked in tests).
Broadcast structured JSON messages over WebSockets (e.g. { "event": "task_created", "task": {...} }).
Achieve 100% test coverage for /auth, /tasks, /admin, and /ws.

🧠 Recap¶

In this lesson, you:

Added notifications via background tasks.
Implemented real-time task updates with WebSockets.
Achieved full test coverage using pytest and TestClient.