Build a packaged CLI and FastAPI service with a transactional database, data imports, and layered tests.

Capstone: civic data tracker

Build a small system that imports public service requests, stores normalized records, summarizes them, and exposes the same application behavior through a CLI and HTTP API.

The project is intentionally larger than one script. Its purpose is to practice boundaries: packaging, validation, database transactions, HTTP behavior, numerical summaries, tabular imports, and testing strategy.

Product brief

Users need to:

import service requests from CSV;
add and update requests manually;
list requests by status, category, and date range;
view category-level counts and resolution times;
use either a CLI or a JSON API;
repeat an import without creating duplicates;
understand rejected rows without leaking sensitive data.

Each request has:

an external ID;
category and description;
creation and optional resolution timestamps;
status: open, in_progress, or resolved;
optional latitude and longitude;
source metadata.

Acceptance criteria

The first release is complete when:

uv sync --locked creates a working environment from a clean clone.
uv run civic --help and uv run fastapi run ... start the two interfaces.
imports are transactional and idempotent by external ID;
invalid rows produce a bounded, actionable report;
SQL values are parameterized and schema constraints protect invariants;
API input and output have separate typed models;
all network calls have explicit timeouts and domain errors;
unit and integration suites pass with branch coverage;
a built wheel installs and runs in a clean environment.

Suggested structure

civic-tracker/
├── pyproject.toml
├── uv.lock
├── README.md
├── src/
│   └── civic_tracker/
│       ├── __init__.py
│       ├── api.py
│       ├── cli.py
│       ├── config.py
│       ├── domain/
│       │   ├── models.py
│       │   └── services.py
│       ├── application/
│       │   ├── commands.py
│       │   └── ports.py
│       ├── adapters/
│       │   ├── database.py
│       │   ├── imports.py
│       │   └── source_client.py
│       └── analytics.py
└── tests/
    ├── unit/
    ├── integration/
    └── e2e/

Keep the CLI and API thin. Both should call the same application services rather than duplicating validation or SQL.

Bootstrap the package

uv init --package civic-tracker
cd civic-tracker
uv python pin 3.12
uv add "fastapi[standard]" httpx sqlalchemy pandas numpy
uv add --dev pytest pytest-cov ruff mypy

Add the entry point:

[project.scripts]
civic = "civic_tracker.cli:main"

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = [
  "-ra",
  "--strict-markers",
  "--cov=civic_tracker",
  "--cov-branch",
  "--cov-report=term-missing",
]
markers = [
  "integration: uses a real adapter",
  "e2e: exercises an installed public interface",
]

Milestone 1: model the domain

Create immutable domain values independent of FastAPI and SQLAlchemy:

from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum


class RequestStatus(StrEnum):
    OPEN = "open"
    IN_PROGRESS = "in_progress"
    RESOLVED = "resolved"


@dataclass(frozen=True, slots=True)
class ServiceRequest:
    external_id: str
    category: str
    description: str
    status: RequestStatus
    created_at: datetime
    resolved_at: datetime | None = None

    def resolution_seconds(self) -> float | None:
        if self.resolved_at is None:
            return None
        duration = (self.resolved_at - self.created_at).total_seconds()
        if duration < 0:
            raise ValueError("resolved_at cannot precede created_at")
        return duration

Decide and test:

whether status can move backward;
whether resolved records require resolved_at;
whether external IDs are case-sensitive;
how timestamps without time zones are handled;
maximum lengths and allowed empty values.

Use aware UTC datetimes internally. Preserve source timezone only if the product needs it.

Milestone 2: design persistence

Start with SQLite and SQLAlchemy 2.x. Define database models separately from domain dataclasses.

Minimum schema:

CREATE TABLE service_requests (
    id INTEGER PRIMARY KEY,
    external_id TEXT NOT NULL UNIQUE,
    category TEXT NOT NULL,
    description TEXT NOT NULL,
    status TEXT NOT NULL
        CHECK (status IN ('open', 'in_progress', 'resolved')),
    created_at TEXT NOT NULL,
    resolved_at TEXT,
    source TEXT NOT NULL,
    CHECK (resolved_at IS NULL OR resolved_at >= created_at)
);

CREATE INDEX idx_service_requests_status
    ON service_requests(status);

CREATE INDEX idx_service_requests_category_created
    ON service_requests(category, created_at);

Create a repository protocol:

from collections.abc import Sequence
from typing import Protocol


class RequestRepository(Protocol):
    def get_by_external_id(self, external_id: str) -> ServiceRequest | None: ...

    def upsert_many(self, requests: Sequence[ServiceRequest]) -> int: ...

    def list(self, filters: RequestFilters) -> list[ServiceRequest]: ...

The application owns transaction intent; the adapter owns SQL mechanics. For a batch import, use one transaction so a failed row cannot leave an unexplained partial import. If product requirements allow partial success, make it an explicit mode with per-row outcomes.

Write integration tests for:

uniqueness and check constraints;
rollback after a mid-batch failure;
parameterized filtering;
ordering and pagination;
migration from an empty database.

Milestone 3: build the CSV pipeline

Separate stages:

read required columns with explicit dtypes;
normalize column names and identifiers;
parse timestamps as UTC;
detect invalid and duplicate rows;
map valid rows to domain objects;
persist in one application operation;
return a summary.

Define an outcome:

@dataclass(frozen=True, slots=True)
class ImportSummary:
    rows_read: int
    rows_accepted: int
    rows_rejected: int
    rows_created: int
    rows_updated: int

Do not let Pandas objects cross into every layer. Convert validated records to domain values at the import boundary. Bound error samples—for example, return the first 20 errors plus the total count.

Practice files should include:

valid data;
duplicate external IDs;
missing IDs;
malformed timestamps;
resolution before creation;
unknown statuses;
identifiers with leading zeroes.

Milestone 4: add analytics

Pandas can produce grouped operational summaries:

def category_summary(frame: pd.DataFrame) -> pd.DataFrame:
    required = {"external_id", "category", "created_at", "resolved_at"}
    missing = required.difference(frame.columns)
    if missing:
        raise ValueError(f"missing columns: {sorted(missing)}")

    working = frame.copy()
    working["resolution_hours"] = (
        working["resolved_at"] - working["created_at"]
    ).dt.total_seconds() / 3600

    return (
        working.groupby("category", as_index=False)
        .agg(
            request_count=("external_id", "nunique"),
            resolved_count=("resolved_at", "count"),
            median_resolution_hours=("resolution_hours", "median"),
        )
        .sort_values("request_count", ascending=False)
        .reset_index(drop=True)
    )

Use NumPy for numerical operations where arrays are natural, such as percentiles:

def percentiles(values: np.ndarray) -> dict[str, float]:
    clean = values[np.isfinite(values)]
    if clean.size == 0:
        raise ValueError("at least one finite value is required")
    p50, p90, p99 = np.percentile(clean, [50, 90, 99])
    return {"p50": float(p50), "p90": float(p90), "p99": float(p99)}

Define the policy for unresolved requests and invalid durations. Test shapes, dtypes, missing values, and expected tolerances.

Milestone 5: build the CLI

Keep main importable and return an exit code:

import argparse
from collections.abc import Sequence


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="civic")
    subcommands = parser.add_subparsers(dest="command", required=True)

    import_parser = subcommands.add_parser("import")
    import_parser.add_argument("path")

    list_parser = subcommands.add_parser("list")
    list_parser.add_argument("--status", choices=[status.value for status in RequestStatus])

    subcommands.add_parser("summary")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        return dispatch(args)
    except UserInputError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    except ApplicationError as exc:
        print(f"operation failed: {exc}", file=sys.stderr)
        return 1

Add if __name__ == "__main__": raise SystemExit(main()) for module execution. Keep expected user errors concise; unexpected errors should retain diagnostics in logs.

Test parser behavior, output, and exit codes without subprocesses. Add one wheel-level E2E test that runs the installed civic command.

Milestone 6: expose a typed API

Endpoints:

GET /health/live
GET /health/ready
POST /requests
GET /requests/{external_id}
GET /requests with bounded filters and pagination
PATCH /requests/{external_id}
POST /imports for bounded uploads or server-side import jobs
GET /analytics/categories

Define separate Pydantic create, patch, and response models. Use dependency injection for application services and database sessions. Return:

201 for successful creation;
404 for an unknown external ID;
409 for uniqueness or invalid state transitions;
422 for structural request validation;
503 when readiness dependencies are unavailable.

For large imports, do not hold an HTTP request open indefinitely. Store the upload safely, enqueue a job, return 202, and expose job status. The first capstone version may impose a small upload limit and process synchronously.

Milestone 7: integrate a remote source

Use HTTPX with:

one lifecycle-managed client;
explicit connect, read, write, and pool timeouts;
validated response models;
pagination limits;
retries only for safe transient failures;
sanitized logs;
a domain exception hierarchy.

Add adapter tests with httpx.MockTransport, then one opt-in contract test against a controlled sandbox. Never make the default unit suite depend on public internet.

Milestone 8: create the test portfolio

Unit tests:

domain state transitions and timestamp rules;
row normalization and mapping;
CLI dispatch;
retry decision policy;
NumPy percentile behavior;
Pandas summaries and join cardinality.

Integration tests:

repository queries and transactions;
SQLite constraints and migrations;
FastAPI routes through TestClient;
CSV-to-database import;
HTTP adapter serialization through a controlled transport.

E2E tests:

install the wheel, import a fixture through the CLI, then query it;
start the API with a temporary database and complete one create/read flow.

Enable branch coverage:

uv run pytest --cov=civic_tracker --cov-branch --cov-report=term-missing

Set a baseline only after inspecting meaningful gaps. Require changed business logic to include tests even when repository-wide coverage remains above threshold.

Milestone 9: harden and ship

Run:

uv run ruff format --check .
uv run ruff check .
uv run mypy src
uv run pytest
uv build
uvx twine check dist/*

Before release:

store configuration in environment variables or a config file, never source;
validate configuration at startup with clear non-secret errors;
use migrations for schema evolution;
back up persistent databases and test restore;
configure bounded logging and correlation IDs;
document concurrency limits and SQLite tradeoffs;
test shutdown so clients and connections close;
generate and inspect OpenAPI;
install the wheel in a clean environment.

Stretch goals

Replace SQLite with PostgreSQL and run the same repository contract suite.
Add an async HTTP adapter while keeping database behavior coherent.
Queue large imports and expose job progress.
Export Parquet summaries with explicit schemas.
Add authentication and resource-level authorization.
Add idempotency keys for API writes.
Profile a large import before optimizing vectorized transforms.

Final review questions

Can the core application run without FastAPI or argparse?
Does one business operation map to one transaction?
Are all SQL values parameterized?
Are external payloads validated before entering the domain?
Are HTTP timeouts, retries, and errors explicit?
Do analytics preserve key uniqueness and join cardinality?
Do tests use real adapters where mocks would give false confidence?
Can a new contributor reproduce the environment from committed metadata?
Does the built artifact work outside the repository?

If every answer is supported by a test, command, or documented design decision, the capstone is more than a demo: it is a maintainable small system.

Capstone — civic data tracker