The Operator Console

The intake pipeline works. The database fills every morning. And Maria's workflow is still: open the SQLite file, squint at forty rows, and decide which ones need a crew today — from memory. Data becomes decisions only when there's somewhere to make the decisions. This week: the operator console. Deliberately un-fancy, ruthlessly practical — a queue, filters, a detail view, actions, and an audit trail for every click.

Assumes: Posts 1–9 and Milestone 2 (the intake DB this console reads). Installs: pip install fastapi httpx python-multipart uvicorn.

Before you code: clarify the ask

Input: the intake DB from Milestone 2 (the service_requests table)
Output: a browser page Maria opens at 8 AM: the queue, filters, request detail, three actions, and an audit trail
Deadline: ops starts using it Monday; Lisa audits the trail the Monday after

Notice what Maria didn't ask for: charts, a dashboard, real-time anything. She asked for a queue and proof. Build the queue and the proof; the charts can wait forever.

The minimal concept

Three ideas, and the first one is the whole lesson:

Operators need a queue with decisions, not a dashboard with charts. A dashboard answers "how are we doing." An operator's morning question is "what do I do first" — that's a sorted, filterable list of work items, each one openable, each one actionable. If your UI can't dispatch a crew, it's a report wearing a costume.

Server-rendered HTML is enough. The browser already knows how to render tables and submit forms — no JavaScript framework, no build step, nothing for Dev to deploy except one Python file. The workflow is the product; the pixels are a rounding error. (This is also why the lesson's pages are plain HTML strings: honest about what they are.)

Every action is a row. Escalate, assign, resolve — each one writes a timestamped row to an audit log: who, what, which request, what note. The audit trail isn't a feature you add later; it's the feature Lisa audits. Build it first and the rest is just UI.

Build it, part 1: two tables

The console needs its own working tables. The intake pipeline owns service_requests; the console adds a requests work-queue view of it plus the audit log. For the lesson we build both tables standalone so the console runs on its own:

import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS requests (
  unique_key     TEXT PRIMARY KEY,
  created_at     TEXT NOT NULL,
  agency         TEXT NOT NULL,
  complaint_type TEXT NOT NULL,
  descriptor     TEXT,
  borough        TEXT,
  status         TEXT NOT NULL DEFAULT 'Open',
  priority       TEXT NOT NULL DEFAULT 'normal'
);
CREATE TABLE IF NOT EXISTS audit_log (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  ts         TEXT NOT NULL,
  actor      TEXT NOT NULL,
  action     TEXT NOT NULL,
  unique_key TEXT NOT NULL,
  note       TEXT
);
"""

SEED = [
    ("10000001", "2026-09-28T08:12:00", "DSNY", "Missed Collection",
     "Garbage not collected on scheduled day", "BROOKLYN", "Open", "normal"),
    ("10000004", "2026-09-29T11:22:00", "HPD", "HEAT/HOT WATER",
     "No heat in apartment 4B", "BRONX", "Open", "urgent"),
    ("10000009", "2026-10-02T09:02:00", "HPD", "HEAT/HOT WATER",
     "Hot water intermittent since Monday", "BROOKLYN", "Escalated", "urgent"),
    # ... 9 more rows, same shape: 12 total across 5 boroughs
]

conn = sqlite3.connect("console.db")
conn.executescript(SCHEMA)
conn.executemany("INSERT OR IGNORE INTO requests VALUES (?,?,?,?,?,?,?,?)", SEED)
conn.commit()
print(f"seeded {len(SEED)} requests")

seeded 12 requests

The seed rows are labeled as seed — realistic, not real feed rows. Twelve complaints across five boroughs, one already escalated, one in progress: enough texture to make the filters and the queue feel real.

Build it, part 2: the queue

The queue is one route. Filters are query parameters, and — Post 3's lesson applied — the values are data, never SQL:

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()

def db():
    conn = sqlite3.connect("console.db")
    conn.row_factory = sqlite3.Row
    return conn

@app.get("/", response_class=HTMLResponse)
def queue(borough: str = "", agency: str = "", status: str = ""):
    where, params = [], []
    for col, val in (("borough", borough), ("agency", agency), ("status", status)):
        if val:
            where.append(f"{col} = ?")   # column names are ours; values are bound
            params.append(val)
    sql = "SELECT * FROM requests"
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY created_at DESC"
    with db() as conn:
        rows = conn.execute(sql, params).fetchall()
        # ... dropdown options + HTML table rows, one <tr> per request
    return HTML_PAGE

Verified with Post 7's TestClient pattern — no browser needed, the HTML is just asserted:

from fastapi.testclient import TestClient
client = TestClient(app)

r = client.get("/")
print("GET / ->", r.status_code, "| rows in queue:", r.text.count("<tr>") - 1)

r = client.get("/", params={"borough": "QUEENS"})
print("GET /?borough=QUEENS ->", r.status_code, "| rows:", r.text.count("<tr>") - 1)

GET / -> 200 | rows in queue: 12
GET /?borough=QUEENS -> 200 | rows: 3

Twelve in the queue, three for Queens — the real seed, the real filter. Maria's 8 AM question ("what's open in my borough?") is now a URL.

Build it, part 3: detail and the action

Clicking a row opens the detail view: the full complaint, its history, and the only three things Maria can do — escalate, assign, resolve. The form is plain HTML; the route is where the discipline lives:

from datetime import datetime, timezone
from fastapi import Form
from fastapi.responses import RedirectResponse

@app.get("/requests/{unique_key}", response_class=HTMLResponse)
def detail(unique_key: str):
    with db() as conn:
        r = conn.execute(
            "SELECT * FROM requests WHERE unique_key = ?", (unique_key,)
        ).fetchone()
        history = conn.execute(
            "SELECT ts, actor, action, note FROM audit_log "
            "WHERE unique_key = ? ORDER BY id DESC",
            (unique_key,),
        ).fetchall()
    # ... render fields, the action form, and the history list
    return HTML_PAGE

# The one rule this app never breaks: the status change and the audit row
# are written in a single transaction. One commit, or neither happens.
@app.post("/requests/{unique_key}/actions")
def record_action(unique_key: str, actor: str = Form(...),
                   action: str = Form(...), note: str = Form("")):
    new_status = {"escalate": "Escalated",
                  "assign": "In Progress",
                  "resolve": "Closed"}[action]
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with db() as conn:  # commit on clean exit, rollback on exception
        conn.execute(
            "UPDATE requests SET status = ? WHERE unique_key = ?",
            (new_status, unique_key))
        conn.execute(
            "INSERT INTO audit_log (ts, actor, action, unique_key, note) "
            "VALUES (?,?,?,?,?)",
            (now, actor, action, unique_key, note))
    return RedirectResponse(f"/requests/{unique_key}", status_code=303)

Maria escalates the Bronx heat complaint. The real run, end to end:

r = client.post("/requests/10000004/actions",
                data={"actor": "maria", "action": "escalate",
                      "note": "crew dispatched, ticket HPD-881"})
row = conn.execute("SELECT ts, actor, action, unique_key, note "
                   "FROM audit_log").fetchone()
print("audit_log row:", tuple(row))
print("request now:", conn.execute(
    "SELECT status, priority FROM requests "
    "WHERE unique_key='10000004'").fetchone())

audit_log row: ('2026-10-03T08:00:31+00:00', 'maria', 'escalate', '10000004', 'crew dispatched, ticket HPD-881')
request now: ('Escalated', 'urgent')

One POST, one transaction: the status moved to Escalated and the audit row exists with Maria's name, a UTC timestamp, and her note. "Prove what we did" — that's the row.

Build it, part 4: the audit trail

The page Lisa will actually open. Every recorded action, newest first, no filters to hide behind:

@app.get("/audit", response_class=HTMLResponse)
def audit():
    with db() as conn:
        rows = conn.execute(
            "SELECT ts, actor, action, unique_key, note FROM audit_log "
            "ORDER BY id DESC").fetchall()
    # ... one <tr> per action: time (UTC), who, action, request, note
    return HTML_PAGE

GET /audit -> 200 | mentions escalate: True

When Lisa asks "who escalated complaint 10000004 and when?", the answer is a row in a table, not a memory. That row is the whole reason the console exists.

Break it, three ways

1. The GET that writes. The action form uses POST — deliberately. Watch what happens when the escalate is a GET link instead (reproduced on purpose with a throwaway route):

@bad.get("/escalate/{uk}")
def bad_escalate(uk: str):
    with sqlite3.connect("console.db") as c:
        c.execute("INSERT INTO audit_log (ts, actor, action, unique_key) "
                  "VALUES ('now','maria','escalate',?)", (uk,))

two GETs, one human decision -> audit rows for 10000003: 2

Two GETs — a double-click, a prefetch, a crawler — and the audit trail claims Maria escalated twice. GETs are for reading; anything that changes the world goes through POST. Post 6's retry rule said the same thing from the client side; the server side is stricter.

2. The helpful filter. The queue builds its WHERE clause with an f-string — the "I'll parameterize it later" version. Feed it borough="' OR '1'='1":

f-string filter with "' OR '1'='1" -> 12 rows (whole table leaked)
parameterized filter with "' OR '1'='1" -> 0 rows

The f-string version returns all twelve rows — the filter is decoration, and Maria is looking at data she didn't ask for. The parameterized version from part 2 treats the input as a borough name that matches nothing: 0 rows. Same input, same database, different trust. The column names in the f-string are safe (they're ours, from a fixed list); the values are never string-interpolated. That's the whole rule.

3. The status without the audit. The nastiest break: the status update commits, then the process dies before the audit insert. Split writes, reproduced on purpose:

status says 'Escalated', audit rows: 0  <- the lie

The queue shows Escalated; the audit trail shows nothing. Maria did the work and has no proof — the exact failure the console was built to prevent. The fix is the single transaction in part 3: the with db() as conn: block commits on clean exit and rolls back on exception. Status and audit are one atomic fact, or neither exists.

Productionize: one file, one command

The console runs as-is. One file, one command, no build step — exactly what Dev asked for:

$ python console.py   # uvicorn inside, port 8123
$ curl -s "http://127.0.0.1:8123/?borough=BROOKLYN" | grep -o "<tr>.\{0,120\}" | head -5
<tr><th>id</th><th>borough</th><th>agency</th><th>type</th><th>status</th><th>priority</th></tr>
<tr><td><a href="/requests/10000012">10000012</a></td><td>BROOKLYN</td><td>DOT</td><td>Streetlight Condition</td><td>Open</t
<tr><td><a href="/requests/10000009">10000009</a></td><td>BROOKLYN</td><td>HPD</td><td>HEAT/HOT WATER</td><td>Escalated</td>
<tr><td><a href="/requests/10000005">10000005</a></td><td>BROOKLYN</td><td>DOB</td><td>Illegal Construction</td><td>Open</td></td></td></td>
<tr><td><a href="/requests/10000001">10000001</a></td><td>BROOKLYN</td><td>DSNY</td><td>Missed Collection</td><td>Open</td></td></td></td>

queue page: HTTP 200, 2548 bytes

Real HTTP, real rows, served from the same file the tests exercised — the lines cut off mid-tag because grep -o was told to show 120 characters, not because the page is broken. Four production notes, stated honestly:

Logging. uvicorn logs to the terminal; Post 8's file-handler habit applies unchanged — the console's own actions are already in audit_log, but server errors need a log file Maria can hand you.

Auth: not yet, and that's disclosed. This v1 has no login — anyone on the network can escalate. Lisa's audit trail records who the form claimed, which is accountability theater without authentication. The login layer is a later lesson in this stage; shipping v1 to the office network is fine, shipping it to the internet is not. Say that out loud before Monday.

One database, two writers. The intake pipeline writes service_requests; the console reads it and writes statuses back. SQLite handles this fine at ops-team scale with one rule: short transactions, which the console already has. If the team grows past a handful of concurrent users, that's the day to revisit — not before.

Backups. The audit trail is evidence now. The DB file joins the backup set alongside the intake database — an audit log you can't restore is a story, not proof.

Explain it to the customer

The caveat is doing the most important work in that paragraph. An operator console without auth is useful and dangerous in the same breath — telling Maria exactly which one it is, and where the boundary lies, is what keeps it on the useful side.

Where this lands in CityOps

The console is the intake pipeline's other half. Milestone 2 built the machine that fills the database every morning with proof; the console is where a human meets that database and makes decisions — with proof. The audit_log table joins the pipeline's audit_log from Milestone 2 as the second half of the evidence story: the pipeline proves what the data did, the console proves what the people did. In Milestone 3 this becomes the operator console v1 proper — with login, exports, and the webhook alerts feeding straight into the queue.

Post 6's principle was don't trust the happy path. The console's: operators don't need dashboards. They need a queue with decisions — and every decision written down, because the decision you can't prove is the one Lisa asks about.