End-to-end Verification, README, and Portfolio

Overview

The code is complete. Now come the three things that separate a project that "works" from one that's "portfolio-worthy": a verification script that exercises every endpoint, a professional README any recruiter can read in 60 seconds, and tests with TestClient that prove your architecture is testable.

This capsule also closes the whole guide — it bridges to PostgreSQL & SQLAlchemy (guide #8), where the in-memory tasks_db gets replaced by a real database, and to Authentication (guide #9), where the assignee field gets connected to authenticated users.

---

The final structure

task-manager-api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── data.py
│   ├── dependencies.py
│   ├── websocket_manager.py
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── tasks.py
│   │   ├── uploads.py
│   │   └── websocket.py
│   ├── middleware/
│   │   ├── __init__.py
│   │   └── logging.py
│   ├── background/
│   │   ├── __init__.py
│   │   └── notifications.py
│   └── uploads/                  # created at runtime
├── tests/
│   ├── __init__.py
│   └── test_api.py
├── .gitignore
├── requirements.txt
└── README.md

14 Python files + 4 configuration files. Each module has a single purpose — that's what keeps the project clean as it grows.

---

requirements.txt and setup

Create requirements.txt:

fastapi>=0.109.0
uvicorn[standard]>=0.27.0
python-multipart>=0.0.9
httpx>=0.26.0
pytest>=8.0.0

Create .gitignore:

__pycache__/
*.py[cod]
*.egg-info/
.pytest_cache/
.venv/
venv/
.env
app/uploads/*
!app/uploads/.gitkeep

Setup from scratch:

# Clone / enter the project
cd task-manager-api

# Create the venv
python -m venv .venv
source .venv/bin/activate  # Linux/macOS
# .venv\Scripts\activate   # Windows

# Install
pip install -r requirements.txt

# Create .gitkeep to keep the uploads folder in git without committing files
mkdir -p app/uploads
touch app/uploads/.gitkeep

# Start it
uvicorn app.main:app --reload

If everything is fine, you see:

INFO:     Uvicorn running on http://127.0.0.1:8000
INFO     [task_manager] Task Manager API started with sample data
INFO:     Application startup complete.

Open http://localhost:8000/docs and you should see Swagger UI with the endpoints organized by tags (Tasks, Uploads, System).

---

End-to-end verification

Create a verify.sh script (or run the commands by hand):

#!/bin/bash
set -e

BASE="http://localhost:8000"

echo "=== 1. Health check ==="
curl -s "$BASE/health" | python -m json.tool

echo "=== 2. List tasks (seed data) ==="
curl -s "$BASE/tasks/" | python -m json.tool

echo "=== 3. Create a task ==="
NEW_TASK=$(curl -s -X POST "$BASE/tasks/" \
  -H "Content-Type: application/json" \
  -d '{"title": "E2E test", "priority": "high", "tags": ["e2e", "test"]}')
echo "$NEW_TASK" | python -m json.tool
TASK_ID=$(echo "$NEW_TASK" | python -c "import sys, json; print(json.load(sys.stdin)['id'])")
echo "→ Created task ID: $TASK_ID"

echo "=== 4. Filter by status ==="
curl -s "$BASE/tasks/?status=pending&limit=3" | python -m json.tool

echo "=== 5. Search by text ==="
curl -s "$BASE/tasks/?search=e2e" | python -m json.tool

echo "=== 6. Detail ==="
curl -s "$BASE/tasks/$TASK_ID" | python -m json.tool

echo "=== 7. PATCH to completed ==="
curl -s -X PATCH "$BASE/tasks/$TASK_ID" \
  -H "Content-Type: application/json" \
  -d '{"status": "completed"}' | python -m json.tool

echo "=== 8. Stats ==="
curl -s "$BASE/tasks/stats" | python -m json.tool

echo "=== 9. CSV export ==="
curl -s "$BASE/tasks/export/csv" -o /tmp/tasks_export.csv
echo "First lines of the CSV:"
head -3 /tmp/tasks_export.csv

echo "=== 10. Upload attachment ==="
echo "Test attachment content" > /tmp/upload_test.txt
ATTACHMENT=$(curl -s -X POST "$BASE/tasks/$TASK_ID/attachments" \
  -F "file=@/tmp/upload_test.txt" \
  -F "description=E2E verification")
echo "$ATTACHMENT" | python -m json.tool
FILENAME=$(echo "$ATTACHMENT" | python -c "import sys, json; print(json.load(sys.stdin)['filename'])")

echo "=== 11. List attachments ==="
curl -s "$BASE/tasks/$TASK_ID/attachments" | python -m json.tool

echo "=== 12. Download attachment ==="
curl -s "$BASE/tasks/$TASK_ID/attachments/$FILENAME" -o /tmp/downloaded.txt
diff /tmp/upload_test.txt /tmp/downloaded.txt && echo "✓ Files are identical"

echo "=== 13. Validation: type not allowed ==="
echo "<xml/>" > /tmp/bad.xml
curl -s -X POST "$BASE/tasks/$TASK_ID/attachments" \
  -F "file=@/tmp/bad.xml;type=application/xml" \
  -w "\nHTTP: %{http_code}\n"

echo "=== 14. DELETE the task ==="
curl -s -X DELETE "$BASE/tasks/$TASK_ID" -w "HTTP: %{http_code}\n"

echo "=== 15. Check the timing header ==="
curl -s -i "$BASE/health" | grep -i "x-process-time"

echo ""
echo "✅ E2E verification complete"

Run it:

chmod +x verify.sh
./verify.sh

You should see 15 sections that finish correctly and E2E verification complete at the end. If any step fails with an error, check the server logs.

WebSocket E2E

In one terminal:

websocat ws://localhost:8000/ws

In another:

curl -s -X POST http://localhost:8000/tasks/ \
  -H "Content-Type: application/json" \
  -d '{"title": "WS test", "priority": "low"}'

In the first terminal you'll see the task_created event. If this works, all 6 features are integrated correctly.

---

Tests with TestClient

Create tests/__init__.py (empty) and tests/test_api.py:

import io
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.data import seed_data, tasks_db


@pytest.fixture(autouse=True)
def reset_data():
    """Resets the data before each test."""
    seed_data()
    yield
    tasks_db.clear()


@pytest.fixture
def client():
    return TestClient(app)


def test_health_check(client):
    response = client.get("/health")
    assert response.status_code == 200
    body = response.json()
    assert body["status"] == "healthy"
    assert "tasks_count" in body


def test_list_tasks_with_seed_data(client):
    response = client.get("/tasks/")
    assert response.status_code == 200
    tasks = response.json()
    assert len(tasks) == 5
    assert all("id" in t and "title" in t for t in tasks)


def test_create_task(client):
    payload = {
        "title": "Test task",
        "priority": "high",
        "tags": ["test"]
    }
    response = client.post("/tasks/", json=payload)
    assert response.status_code == 201
    body = response.json()
    assert body["title"] == "Test task"
    assert body["priority"] == "high"
    assert body["status"] == "pending"
    assert body["id"] == 6


def test_get_task_not_found(client):
    response = client.get("/tasks/9999")
    assert response.status_code == 404
    body = response.json()
    assert "detail" in body


def test_filter_by_priority(client):
    response = client.get("/tasks/?priority=critical")
    assert response.status_code == 200
    tasks = response.json()
    assert all(t["priority"] == "critical" for t in tasks)


def test_patch_to_completed_triggers_completion(client):
    response = client.patch("/tasks/3", json={"status": "completed"})
    assert response.status_code == 200
    assert response.json()["status"] == "completed"


def test_stats_returns_aggregates(client):
    response = client.get("/tasks/stats")
    assert response.status_code == 200
    stats = response.json()
    assert stats["total"] == 5
    assert "by_status" in stats
    assert "completion_rate" in stats


def test_csv_export(client):
    response = client.get("/tasks/export/csv")
    assert response.status_code == 200
    assert "text/csv" in response.headers["content-type"]
    assert "attachment" in response.headers["content-disposition"]
    assert "id,title" in response.text


def test_upload_invalid_content_type(client):
    files = {"file": ("test.xml", io.BytesIO(b"<xml/>"), "application/xml")}
    response = client.post("/tasks/1/attachments", files=files)
    assert response.status_code == 415


def test_upload_valid_file(client):
    files = {"file": ("notes.txt", io.BytesIO(b"hello world"), "text/plain")}
    response = client.post(
        "/tasks/1/attachments",
        files=files,
        data={"description": "Test upload"}
    )
    assert response.status_code == 201
    body = response.json()
    assert body["original_name"] == "notes.txt"
    assert body["size_bytes"] == 11


def test_websocket_connect_and_ping(client):
    with client.websocket_connect("/ws") as ws:
        welcome = ws.receive_json()
        assert welcome["type"] == "connection_established"

        ws.send_json({"action": "ping", "data": {"hello": "world"}})
        pong = ws.receive_json()
        assert pong["type"] == "pong"


def test_middleware_adds_timing_header(client):
    response = client.get("/health")
    assert "x-process-time-ms" in {k.lower() for k in response.headers}

Run it:

pytest tests/ -v

You should see 12 tests passing. These aren't exhaustive tests — they're demonstrations that the architecture is testable: the TestClient reaches the app like a normal HTTP client, the dependencies get injected, and seed_data() gives you a known state in every test.

Why the tests are short

If you had the data hardcoded inside each endpoint (no DI, no tasks_db extracted), you'd have to mock every call. Because you did DI properly back in capsule 02, the tests are readable and fast.

This is what module 1 (Dependency Injection) really means: code you can test without fighting the framework.

---

README.md

Create README.md:

# Task Manager API

An advanced REST API for task management, built with FastAPI. It includes dependency injection, modular routers, WebSockets for real-time notifications, file uploads, background tasks, and custom middleware.

## Features

- **Full task CRUD** with pagination, filters, and search
- **WebSocket** (`/ws`) for real-time events
- **File uploads** with type and size validation
- **Background tasks** for email notifications and an audit log
- **CSV export** via StreamingResponse
- **Middleware** for automatic logging and timing
- **Stats endpoint** with aggregations
- **OpenAPI documentation** at `/docs`

## Tech Stack

- Python 3.9+
- FastAPI 0.109+
- Pydantic v2
- uvicorn (ASGI server)
- pytest + httpx (testing)

## Setup

```bash
git clone <your-repo>
cd task-manager-api

python -m venv .venv
source .venv/bin/activate

pip install -r requirements.txt

uvicorn app.main:app --reload

Open http://localhost:8000/docs.

Quick Start

Create a task

curl -X POST http://localhost:8000/tasks/ \
  -H "Content-Type: application/json" \
  -d '{"title": "My first task", "priority": "high"}'

Upload a file

curl -X POST http://localhost:8000/tasks/1/attachments \
  -F "file=@./document.pdf" \
  -F "description=Project specs"

Connect to the WebSocket

websocat ws://localhost:8000/ws

API Reference

Method	Endpoint	Description
GET	/health	Health check
POST	/tasks/	Create a task
GET	/tasks/	List tasks (filters, pagination)
GET	/tasks/{id}	Detail
PUT	/tasks/{id}	Full update
PATCH	/tasks/{id}	Partial update
DELETE	/tasks/{id}	Delete
GET	/tasks/stats	Statistics
GET	/tasks/export/csv	Export to CSV
POST	/tasks/{id}/attachments	Upload a file
GET	/tasks/{id}/attachments	List attachments
GET	/tasks/{id}/attachments/{file}	Download
WS	/ws	Event stream

See the OpenAPI docs for the full schemas.

WebSocket events

Event	Trigger
task_created	POST /tasks/
task_updated	PUT /tasks/{id}
task_patched	PATCH /tasks/{id}
task_completed	PATCH with status=completed
task_deleted	DELETE /tasks/{id}
attachment_added	POST /tasks/{id}/attachments

Testing

pytest tests/ -v

Structure

app/
├── main.py              # App + middleware + lifespan
├── models.py            # Pydantic schemas
├── dependencies.py      # Shared injectables
├── websocket_manager.py # ConnectionManager
├── routers/             # Endpoints by domain
├── middleware/          # Custom middleware
└── background/          # Background task helpers

Known limitations

• Data is stored in memory. Restart the server and it's gone.
• There's no authentication. Anyone can call any endpoint.
• Uploads are saved to the local disk. There's no automatic cleanup.

These limitations get solved in later projects:

• PostgreSQL + SQLAlchemy replaces the in-memory storage
• JWT + OAuth2 adds authentication and authorization
• S3 / MinIO replaces the filesystem for uploads

License

MIT

---

## Before vs after

| Aspect | FastAPI Fundamentals | FastAPI Advanced (this project) |
|---------|---------------------|----------------------------------|
| Endpoints | 1 file, all together | Modular routers by domain |
| Shared logic | Copy-paste between endpoints | Injected dependencies |
| Communication | HTTP request/response | + real-time WebSocket |
| Heavy work | Blocks the response | Async background tasks |
| Files | Not supported | Upload + download with validation |
| Observability | `print` or nothing | Middleware + a structured audit log |
| Tests | Endpoint by endpoint, manual | TestClient + fixtures + reuse |
| Structure | A single `main.py` | `app/` with separation of concerns |

If you compare the code from the Fundamentals Guide with this project, you'll see that the domain (task management) grew ~3x, but the cognitive complexity per file stays low. **That's what modular code produces: extensibility without chaos.**

---

## Final troubleshooting

### The server starts but `/docs` is empty
Check that you registered the routers in `main.py` with `app.include_router(...)`. Without that, the endpoints don't show up in the docs.

### The tags in `/docs` don't separate the endpoints
Make sure every `APIRouter` has `tags=["..."]`. Without tags, FastAPI groups them all under "default".

### `pytest` fails with `ModuleNotFoundError: No module named 'app'`
Run pytest from the project root (where `requirements.txt` lives), not from `tests/`. Alternatively, add a `conftest.py` or a `pytest.ini` with `pythonpath = .`.

### The audit log grows without limit
In memory, yes — you solve that when you migrate to a database in guide #8. For now, it's fine.

### `python-multipart` isn't found at startup
It's in `requirements.txt`. If you start without installing it, FastAPI loads without it but fails on uploads. Reinstall with `pip install -r requirements.txt`.

### CORS blocks requests from the frontend
The project is open to every origin (`allow_origins=["*"]`). In production, restrict it to specific domains.

### Background tasks don't run in production
`BackgroundTasks` runs in the same process as the API. For long tasks, or ones that must survive restarts, move to Celery or RQ (covered in module 4).

---

## Completion checklist

### Architecture

- [ ] A folder structure with `app/`, `routers/`, `middleware/`, `background/`
- [ ] Each router in its own file with a `prefix` and `tags`
- [ ] `dependencies.py` with `pagination_params`, `get_task_or_404`, `filter_params`, `apply_filters`
- [ ] `models.py` with separate models (`TaskCreate`, `TaskUpdate`, `TaskPatch`, `TaskResponse`, `TaskSummary`, `TaskStats`)
- [ ] Enums for `Status` and `Priority`
- [ ] A `ConnectionManager` with `connect`, `disconnect`, `broadcast`, `broadcast_event`

### HTTP endpoints

- [ ] `GET /health` returns the system status
- [ ] `POST /tasks/` creates a task, fires the background tasks and the WS broadcast
- [ ] `GET /tasks/` lists with filters (`status`, `priority`, `assignee`, `search`, `tag`) and pagination (`skip`, `limit`)
- [ ] `GET /tasks/{id}` returns the detail or a 404
- [ ] `PUT /tasks/{id}` replaces the whole task
- [ ] `PATCH /tasks/{id}` updates only the fields that were sent
- [ ] `DELETE /tasks/{id}` deletes and fires a broadcast
- [ ] `GET /tasks/stats` returns the aggregations
- [ ] `GET /tasks/export/csv` returns a CSV with `Content-Disposition: attachment`

### File uploads

- [ ] `POST /tasks/{id}/attachments` with multipart, validates type and size
- [ ] `GET /tasks/{id}/attachments` lists them
- [ ] `GET /tasks/{id}/attachments/{filename}` downloads with `FileResponse`
- [ ] Validation: allowed types, max 5MB, no empty file
- [ ] `secrets.token_hex` for the filenames
- [ ] Path traversal blocked on download

### WebSocket

- [ ] The `/ws` endpoint connects and sends a welcome message
- [ ] Client actions: `ping`, `status`, `broadcast_test`
- [ ] Broadcast events: `task_created`, `task_updated`, `task_patched`, `task_completed`, `task_deleted`, `attachment_added`
- [ ] `WebSocketDisconnect` is handled without crashing the server

### Background tasks

- [ ] `send_email_notification` runs after the response
- [ ] `log_audit_event` records every CRUD operation
- [ ] `audit_log` is reachable (through `/health` or a dedicated endpoint)

### Middleware

- [ ] `LoggingMiddleware` adds `X-Process-Time-Ms` to every response
- [ ] `CORSMiddleware` is configured
- [ ] Request logs in the console

### Testing

- [ ] At least 10 tests with `TestClient`
- [ ] The tests cover: basic CRUD, filters, validations, WebSocket
- [ ] The tests use a `fixture` to reset the data
- [ ] `pytest tests/ -v` runs without errors

### Documentation

- [ ] `/docs` shows the endpoints organized by tags
- [ ] Every endpoint has a description and examples
- [ ] `README.md` with setup, features, API reference, limitations
- [ ] `requirements.txt` with specific versions
- [ ] `.gitignore` excludes `.venv`, `__pycache__`, `app/uploads/`

---

## Self-assessment rubric

Go back to **capsule 01** and compare your implementation against the 100-point rubric. Be honest:

- **90-100 (Excellent):** Ready for your portfolio. Push it to GitHub with the README.
- **75-89 (Good):** Functional. Identify the 1-2 weak points and fix them before you present it.
- **60-74 (Acceptable):** It works but needs polish. Review modules 1-5, focusing on the areas with the lowest scores.
- **<60:** Go back. It's not a defeat — it's the moment to consolidate before moving on to the next guide.

If your score is 75+, this project **is** a portfolio piece. Don't wait for it to be "perfect" — recruiters value finished projects with documented limitations more than eternal works in progress.

---

## How to present it in your portfolio

### The GitHub repo

- **Name:** `task-manager-api` or something equivalent. Avoid `my-fastapi-project`.
- **Repo description (1 line):** "REST API + WebSockets with FastAPI: DI, background tasks, file uploads, real-time"
- **Topics:** `fastapi`, `python`, `websockets`, `rest-api`, `pydantic`, `pytest`
- **README up top:** The one you just wrote. It should read in 60 seconds.
- **Optional screenshot:** A capture of `/docs` or of a WS client receiving events. It goes at the top of the README.

### On your CV / LinkedIn

> **Task Manager API** — A REST API with FastAPI implementing dependency injection, WebSockets for real-time notifications, background tasks, validated file uploads, and custom middleware. Tests with TestClient and a modular router architecture. [github.com/your-user/task-manager-api]

One line. Concrete. Link to the repo.

### In technical interviews

When they ask you "tell me about a project you've built", this is your answer:

1. **Problem:** "I built a task management API that combines REST + real-time."
2. **Technical decisions:** "I used dependency injection so every endpoint is testable without heavy mocks."
3. **Trade-offs:** "The data lives in memory — that's the main limitation and it's documented in the README."
4. **What I learned:** "The most interesting challenge was integrating WebSocket broadcasts with background tasks without blocking the response."

Documented trade-offs > flawless code. That's senior thinking.

---

## Connection with the next guide: PostgreSQL & SQLAlchemy

This API has an obvious limitation: **the data lives in memory**. You restart uvicorn and it's gone.

In **Guide #8 — PostgreSQL & SQLAlchemy** you're going to:

- Replace `tasks_db: list[dict]` with SQLAlchemy models
- Use dependency injection to inject the DB session (`Depends(get_db)`) — exactly the pattern you already learned here
- Migrate the `audit_log` into a table
- Set up Alembic for migrations

**Architectural spoiler:** You'll barely touch the routers. The endpoints stay the same. Only `data.py` gets swapped for `database.py`, along with the internal queries. **That's what the investment in architecture pays you back.**

And in **Guide #9 — Authentication**, the `assignee: str | None` field that's just a string today gets connected to an authenticated user with JWT — without rewriting the routers.

---

## Final resources

1. [FastAPI in Production](https://fastapi.tiangolo.com/deployment/) — How to deploy this kind of API
2. [Awesome FastAPI](https://github.com/mjhea0/awesome-fastapi) — A curated list of advanced resources
3. [Real Python: FastAPI Testing](https://realpython.com/fastapi-python-web-apis/#testing-fastapi-with-pytest) — Deeper testing patterns
4. [12 Factor App](https://12factor.net/) — Principles for production-ready apps
5. [Twelve-Factor + FastAPI talk](https://fastapi.tiangolo.com/advanced/) — How to apply these principles to the stack
6. [HTTPX docs](https://www.python-httpx.org/) — The HTTP client the TestClient uses under the hood

---

## What's next?

You've finished the **FastAPI Advanced Features Guide**. You built an API that combines dependency injection, modular routers, advanced response models, background tasks, WebSockets, file uploads, and middleware — the 6 features that separate a portfolio API from a "Hello World" API.

**The path continues like this:**

- **Guide #8 — PostgreSQL & SQLAlchemy:** You replace the in-memory data with a real database. The architecture you already have absorbs it without rewriting routers.
- **Guide #9 — Authentication & Authorization:** JWT, OAuth2, dependencies to protect endpoints.
- **Guide #11 — Testing Backend Applications:** You go deeper on TestClient, integration tests, strategic mocking.

Close the laptop, run `git add . && git commit -m "feat: complete advanced FastAPI features"`, push to GitHub, and move on to the next guide with confidence. Your Task Manager is ready for your portfolio.