FastAPI Event API

Project roadmap

1. 01

   Define the event types

   ~1.5h

   Write Pydantic models for each event type (for example user.signed_up, order.placed and app.error), each with its own data fields and validation, combined into a discriminated union on the type field, plus common fields (a client-generated id, occurred_at and source). Write API.md with the endpoints. Done when: the models validate sample events and reject bad ones with clear errors.

2. 02

   Store events

   ~2h

   Create a FastAPI app with SQLAlchemy and SQLite (or PostgreSQL), with an events table storing the common fields as columns and the data as JSON, and a unique constraint on the event id. Add POST /events and POST /events/batch (up to 500 events), which ignore duplicates and return how many were stored and how many were duplicates, with per-item errors for invalid events in a batch. Done when: events are stored, and resending them stores nothing new.

3. 03

   Query with filters and pagination

   ~2h

   Add GET /events with filters (type, source, and from and to times, validated with Pydantic query models) and cursor pagination (a cursor encoding the last event's time and id), with indexes to keep it fast. Add GET /events/{id}. Done when: filters work, and paging through events while new ones arrive never repeats or skips events.

4. 04

   Build reports with pandas

   ~2h

   Add GET /reports/daily?from=&to= that loads the matching events into a pandas DataFrame (with pd.read_sql) and returns counts per type per day and unique users per day, and GET /reports/export.csv that streams the filtered events as CSV. Limit the date range so reports stay fast. Done when: reports return correct numbers for sample data, and the CSV downloads.

5. 05

   Harden the API

   ~1.5h

   Add API key authentication with a dependency, limit request sizes, reject events far in the future, return consistent error responses, and log each request with its duration. Check the OpenAPI documentation at /docs describes every endpoint and model clearly. Done when: unauthenticated and oversized requests are rejected, and the docs are complete.

6. 06

   Test it

   ~2h

   Write pytest tests with FastAPI's TestClient and a test database: each event type valid and invalid, batch handling with duplicates and invalid items, filters and cursor pagination, reports on known data, authentication, and size limits. Done when: pytest passes.

Resources

• DocsFastAPI tutorial
• DocsPydantic: discriminated unions
• DocsSQLAlchemy 2.0 unified tutorial
• Docspandas: read_sql
• DocsFastAPI: testing