Building a Manual Activity API with FastAPI, Pydantic, and SQLAlchemy

The problem: clean history before forecasting

AI Time Traveler is a personal productivity forecasting project. Before it can project anything, it needs a reliable history of daily activity: coding hours, study hours, LeetCode problems, GitHub commits, and custom metrics a user defines later.

This post covers the backend slice that accepts those manual entries. The requirements were intentionally concrete:

• validate values and dates
• accept one entry or a batch
• upsert on user + metric + date
• filter history by date range and metric
• paginate the results
• let users define custom metrics
• keep the current-user dependency swappable until real auth exists

The stack is FastAPI, Pydantic v2, SQLAlchemy 2, and SQLite for local tests. The repository is AI Time Traveler.

The API surface

The endpoints are small:

• POST /api/activities accepts a single object or a list and upserts each entry.
• GET /api/activities filters by inclusive from/to dates and metric name, with limit/offset pagination.
• GET /api/metrics lists built-in and custom metric definitions.
• POST /api/metrics creates a custom metric.

The response for a create is always a list, even when the request was a single object. That keeps the client code simple: normalize the request, then handle one shape on the way back.

Validation belongs at the edge

The activity request schema enforces the rules that do not need a database lookup:

class ActivityCreate(BaseModel): model_config = ConfigDict(extra="forbid") metric: MetricName date: date value: float = Field(ge=0, allow_inf_nan=False) source: ActivitySource = "manual" @field_validator("date") @classmethod def reject_future_date(cls, value: date) -> date: if value > date.today(): raise ValueError("activity date cannot be in the future") return value 

There is one rule the schema cannot own cleanly: metric existence. Built-in metrics are known at import time, but custom metrics are created at runtime, so the database is the source of truth. The route resolves every metric name in one query and returns a 422 before writing anything if any name is unknown.

That split matters. Pydantic handles shape and value errors; the route handles the one piece of domain state that lives in the database.

Bulk upsert without a write per row

The bulk endpoint normalizes the payload, deduplicates by metric and date, loads the existing rows for those keys, then updates or inserts them in one transaction.

The core looks like this:

entries = payload if isinstance(payload, list) else [payload] if not entries: raise HTTPException(status_code=422, detail="At least one activity entry is required") metrics = _resolve_metrics(db, entries) deduped = {(entry.metric, entry.date): entry for entry in entries} metric_ids = {metrics[name].id for name, _ in deduped} dates = {activity_date for _, activity_date in deduped} existing = { (activity.metric_id, activity.date): activity for activity in db.scalars( select(Activity) .options(selectinload(Activity.metric)) .where( Activity.user_id == current_user.id, Activity.metric_id.in_(metric_ids), Activity.date.in_(dates), ) ) } for (metric_name, activity_date), entry in deduped.items(): metric = metrics[metric_name] activity = existing.get((metric.id, activity_date)) if activity is None: activity = Activity( user_id=current_user.id, metric=metric, date=activity_date, value=entry.value, source=entry.source, ) db.add(activity) existing[(metric.id, activity_date)] = activity else: activity.value = entry.value activity.source = entry.source db.commit() 

After the commit, the route reloads the saved rows with selectinload(Activity.metric) so the response includes metric details and server-generated timestamps.

A few details are worth calling out.

First, the dictionary comprehension gives last-value-wins semantics when the same metric and date appear twice in one request. That is what an upsert batch should do, rather than failing on a duplicate key the client accidentally sent.

Second, the implementation does one metrics query, one existing-rows query, and one commit. It is not a single database-level ON CONFLICT statement, but it avoids the obvious N+1 pattern and is easy to reason about across SQLite and PostgreSQL. For a manual-entry endpoint with small batches, that trade-off is fine.

Filtering and pagination

The read endpoint uses a small page envelope:

@router.get("", response_model=ActivityPage) def list_activities( current_user: CurrentUser, db: DbSession, from_: Annotated[date | None, Query(alias="from")] = None, to: Annotated[date | None, Query()] = None, metric: Annotated[str | None, Query(min_length=1, max_length=64)] = None, limit: Annotated[int, Query(ge=1, le=500)] = 100, offset: Annotated[int, Query(ge=0)] = 0, ) -> ActivityPage: if from_ is not None and to is not None and from_ > to: raise HTTPException(status_code=422, detail="'from' must be on or before 'to'") filters = [Activity.user_id == current_user.id] if from_ is not None: filters.append(Activity.date >= from_) if to is not None: filters.append(Activity.date <= to) if metric is not None: filters.append(Activity.metric.has(MetricDefinition.name == metric)) total = db.scalar(select(func.count()).select_from(Activity).where(*filters)) or 0 items = db.scalars( select(Activity) .options(selectinload(Activity.metric)) .where(*filters) .order_by(Activity.date.desc(), Activity.id.desc()) .limit(limit) .offset(offset) ).all() return ActivityPage(items=items, total=total, limit=limit, offset=offset) 

The date bounds are inclusive. The ordering is date DESC, id DESC, which keeps pagination stable when two activities share a date. The response includes the total matching count before limit and offset, so a client can show "page 1 of N" without a second request.

Offset pagination is a deliberate choice here. Personal activity history is small-to-medium data, and the UI benefits from page numbers. If this ever becomes a large, frequently changing feed, cursor pagination would be the better fit.

A current-user dependency you can replace

There is no real authentication yet, so the API uses a fixed development user. The important part is that routes do not import that detail directly:

DEV_USER_EMAIL = "dev@time-traveler.local" DEV_USER_DISPLAY_NAME = "Development User" def get_current_user(db: DbSession) -> User: user = db.scalar(select(User).where(User.email == DEV_USER_EMAIL)) if user is not None: return user user = User(email=DEV_USER_EMAIL, display_name=DEV_USER_DISPLAY_NAME) db.add(user) try: db.commit() except IntegrityError: db.rollback() user = db.scalar(select(User).where(User.email == DEV_USER_EMAIL)) if user is None: raise return user db.refresh(user) return user CurrentUser = Annotated[User, Depends(get_current_user)] 

Every route depends on CurrentUser. Swapping in JWT or session auth later means changing the dependency, not editing the activity and metric handlers.

Tests that exercise the failure paths

The test suite uses the existing in-memory SQLite fixtures and covers the cases that usually break first:

• negative values
• future dates
• unknown metric names
• empty bulk payloads
• single and bulk creates
• upserts on the same user, metric, and date
• duplicate keys in one request
• date and metric filters
• pagination boundaries and invalid ranges
• custom metric creation and duplicate conflicts
• user scoping and the fixed development user

The final run was 39 passing tests, with ruff and black clean.

What I would do differently

The current upsert is readable and database-agnostic, but a production system with concurrent writers should use a database-level INSERT ... ON CONFLICT DO UPDATE to make the operation atomic under contention. The current select-then-update approach is fine for a single-user development slice, not for a high-write API.

The other obvious next step is real authentication. Once requests carry a user identity, custom metrics should probably become user-scoped too; right now the metric definition table is global.

Finally, if the history grows into a large timeline, cursor pagination with a stable (date, id) cursor is the natural upgrade.

Further reading

These posts shaped the approach:

• Cursor Pagination vs Offset Pagination: Which One Should You Use? by Ahmed Omeiza
• FastAPI + Pydantic: A Practical Guide to Data Validation by Moshfiqur Rahman
• Faster Bulk Inserts in PostgreSQL by nicolas.vbgh

Agent session

I saved the implementation session with DevRelay. It includes the schema, route, dependency, and test work:

backend/app/api/schemas/activity.py

Added ActivityCreate (metric name, date, non-negative finite value, source), future-date validator, ActivityRead with expanded MetricRead, and ActivityPage.

backend/app/api/routes/activities.py

Implemented POST /api/activities with single-or-list input, metric resolution, deduplication, upsert, and a 200 list response.

backend/app/api/routes/activities.py

Added GET /api/activities with date/metric filters, pagination validation, total count, and selectinload for metric details.

backend/app/api/routes/metrics.py

Added GET /api/metrics and POST /api/metrics with MetricCreate validation, 201 creation, and 409 duplicate handling.

cd backend && .venv/bin/pytest -q

39 passed, 1 harmless Starlette/httpx deprecation warning