⚡ Swarm Architecture

Buying Profile Explorer Implementation Plan

📅 Published August 12, 2026 ⏱️ 11 min read 🔬 Peer Verified

# Buying Profile Explorer Implementation Plan

> For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: A paginated, sortable, account-number-filterable dashboard panel showing weekly buying profiles grouped account → user → product-line, 20 accounts/page, biggest-buyers-first.

Architecture: A pure param parser (core/table_params.py) clamps/whitelists page/sort/dir (the ORDER BY/OFFSET injection boundary). A FastAPI HTMX route pages the accounts (aggregating user rows), fetches all line rows for the page's accounts, attaches names + cluster labels, and groups in Python for a grouped/indented template with a filter box, sort controls, and a prev/next pager. Single-DB Postgres.

Tech Stack: Python 3.12, FastAPI + Jinja2 + HTMX, psycopg2/PostgreSQL, pytest.

Spec: docs/superpowers/specs/2026-06-11-buying-profile-explorer-design.md

---

File structure

| File | Responsibility | New/Modify | |---|---|---| | core/table_params.py | Pure parse_explorer_params(raw) — clamp page, whitelist sort/dir, resolve safe column names. | Create | | tests/test_table_params.py | Hermetic tests for the parser. | Create | | dashboard/app.py | buying_profile_explorer_panel route (two queries + grouping). | Modify | | dashboard/templates/_buying_profile_explorer.html | Grouped table + filter + sort controls + pager. | Create | | dashboard/templates/home.html | Panel container in the Customers tab. | Modify |

---

Task 1: Pure param parser

Files:

  • Create: core/table_params.py
  • Test: tests/test_table_params.py

  • [ ] Step 1: Write the failing test

Create tests/test_table_params.py:

`python from core.table_params import parse_explorer_params, PAGE_SIZE

def test_defaults_on_empty(): p = parse_explorer_params({}) assert p["page"] == 0 and p["asort"] == "orders" and p["lsort"] == "rank" assert p["dir"] == "desc" and p["country"] == "GB" and p["q"] is None assert p["asort_col"] == "tot_orders" and p["lsort_col"] == "rank_freq"

def test_clamps_and_whitelists(): p = parse_explorer_params({"page": "-4", "asort": "bogus", "lsort": "; DROP", "dir": "sideways", "country": "fr", "q": " 60320990 "}) assert p["page"] == 0 # negative clamped assert p["asort"] == "orders" # unknown -> default assert p["lsort"] == "rank" # injection attempt -> default assert p["dir"] == "desc" # unknown -> default assert p["country"] == "FR" assert p["q"] == "60320990" # trimmed

def test_valid_values_pass_through(): p = parse_explorer_params({"page": "3", "asort": "units", "lsort": "orders", "dir": "asc"}) assert p["page"] == 3 and p["asort"] == "units" and p["asort_col"] == "tot_units" assert p["lsort"] == "orders" and p["lsort_col"] == "orders_count" and p["dir"] == "asc"

def test_column_names_are_safe_identifiers(): # every resolvable column is a bare identifier (no spaces/punctuation) -> ORDER BY safe import re for raw in ({"asort": k} for k in ("orders", "units", "users")): assert re.fullmatch(r"[a-z_]+", parse_explorer_params(raw)["asort_col"]) for raw in ({"lsort": k} for k in ("rank", "orders", "units")): assert re.fullmatch(r"[a-z_]+", parse_explorer_params(raw)["lsort_col"]) `

  • [ ] Step 2: Run to verify it fails

Run: venv\Scripts\python.exe -m pytest tests/test_table_params.py -v Expected: FAIL — ModuleNotFoundError: No module named 'core.table_params'.

  • [ ] Step 3: Implement

Create core/table_params.py:

`python """Pure param parsing for the Buying Profile Explorer table. Clamps page, whitelists sort keys + direction, and resolves them to SAFE column identifiers so the route never interpolates raw user input into ORDER BY / OFFSET.""" from __future__ import annotations

PAGE_SIZE = 20

# user-facing sort key -> safe SQL column identifier ACCOUNT_SORTS = {"orders": "tot_orders", "units": "tot_units", "users": "n_users"} LINE_SORTS = {"rank": "rank_freq", "orders": "orders_count", "units": "units_qty"}

def parse_explorer_params(raw: dict) -> dict: """raw: the request's query params (str-ish values). Returns a dict with clamped/whitelisted values plus resolved asort_col/lsort_col identifiers.""" def _page(v) -> int: try: return max(0, int(v)) except (TypeError, ValueError): return 0

asort = raw.get("asort") if raw.get("asort") in ACCOUNT_SORTS else "orders" lsort = raw.get("lsort") if raw.get("lsort") in LINE_SORTS else "rank" direction = "asc" if raw.get("dir") == "asc" else "desc" country = "FR" if str(raw.get("country") or "").upper() == "FR" else "GB" q = (str(raw.get("q")).strip() if raw.get("q") is not None else "") or None return { "page": _page(raw.get("page")), "asort": asort, "lsort": lsort, "dir": direction, "country": country, "q": q, "asort_col": ACCOUNT_SORTS[asort], "lsort_col": LINE_SORTS[lsort], } `

  • [ ] Step 4: Run to verify it passes

Run: venv\Scripts\python.exe -m pytest tests/test_table_params.py -v Expected: 4 passed.

  • [ ] Step 5: Commit

`bash git add core/table_params.py tests/test_table_params.py git commit -m "feat(explorer): pure param parser (clamp/whitelist, injection guard)" `

Task 2: The panel route

Files:

  • Modify: dashboard/app.py (add a new route; insert it right AFTER the buying_profile_panel handler — search for @app.get("/panel/buying-profile", response_class=HTMLResponse) and place the new route after that function's return / before the next @app.get)

  • [ ] Step 1: Add the route

Insert this complete handler immediately after the existing buying_profile_panel function in dashboard/app.py:

`python @app.get("/panel/buying-profile-explorer", response_class=HTMLResponse) def buying_profile_explorer_panel(request: Request, q: str | None = None, page: int = 0, asort: str = "orders", lsort: str = "rank", dir: str = "desc", country: str = "GB") -> HTMLResponse: """Paginated account -> user -> product-line buying-profile table. Pages by account (20/page), biggest buyers first; filter by account number; sortable line/account columns. Single-DB Postgres.""" from core.table_params import parse_explorer_params, PAGE_SIZE p = parse_explorer_params({"q": q, "page": page, "asort": asort, "lsort": lsort, "dir": dir, "country": country}) direction = "ASC" if p["dir"] == "asc" else "DESC" base = {"p": p, "page": p["page"], "accounts": [], "has_next": False, "no_match": False, "empty": False, "snapshot_week": None}

snap = crm.query("SELECT max(snapshot_week) AS w FROM buying_profile_weekly " "WHERE source_country=%s", (p["country"],)) snapshot_week = snap[0]["w"] if snap else None if snapshot_week is None: return templates.TemplateResponse(request, "_buying_profile_explorer.html", {base, "empty": True}) base["snapshot_week"] = snapshot_week

# Step 1 — page the accounts (aggregate user rows up to account) if p["q"]: acc_rows = crm.query( """ SELECT u.account_number, SUM(b.orders_count) AS tot_orders, SUM(b.units_qty) AS tot_units, COUNT(DISTINCT b.scope_id) AS n_users FROM buying_profile_weekly b JOIN ecom_users u ON u.ecom_user_id=b.scope_id AND u.source_country=b.source_country WHERE b.scope_type='user' AND b.source_country=%s AND b.snapshot_week=%s AND (u.account_number=%s OR u.account_number=LPAD(%s,10,'0')) GROUP BY u.account_number """, (p["country"], snapshot_week, p["q"], p["q"])) has_next = False else: acc_rows = crm.query( f""" SELECT u.account_number, SUM(b.orders_count) AS tot_orders, SUM(b.units_qty) AS tot_units, COUNT(DISTINCT b.scope_id) AS n_users FROM buying_profile_weekly b JOIN ecom_users u ON u.ecom_user_id=b.scope_id AND u.source_country=b.source_country WHERE b.scope_type='user' AND b.source_country=%s AND b.snapshot_week=%s GROUP BY u.account_number ORDER BY {p['asort_col']} {direction} NULLS LAST, u.account_number LIMIT %s OFFSET %s """, (p["country"], snapshot_week, PAGE_SIZE + 1, p["page"] * PAGE_SIZE)) has_next = len(acc_rows) > PAGE_SIZE acc_rows = acc_rows[:PAGE_SIZE]

if not acc_rows: return templates.TemplateResponse(request, "_buying_profile_explorer.html", {base, "no_match": bool(p["q"])}) acc_nos = [r["account_number"] for r in acc_rows]

meta = {r["account_number"]: r for r in crm.query( """ SELECT a.account_number, a.name, l.name AS cluster_name FROM ecom_accounts a LEFT JOIN account_firmographic_community c ON c.account_number=a.account_number AND c.source_country=a.source_country LEFT JOIN firmographic_community_labels l ON l.community_id=c.community_id AND l.source_country=c.source_country WHERE a.account_number = ANY(%s) AND a.source_country=%s """, (acc_nos, p["country"]))}

# Step 2 — all line rows for these accounts lines = crm.query( f""" SELECT u.account_number, b.scope_id AS ecom_user_id, u.user_email, u.user_name, b.product_reference, b.rank_freq, b.rank_qty, b.orders_count, b.units_qty, pr.product_description FROM buying_profile_weekly b JOIN ecom_users u ON u.ecom_user_id=b.scope_id AND u.source_country=b.source_country LEFT JOIN ecom_products pr ON pr.product_reference=b.product_reference AND pr.source_country=b.source_country WHERE b.scope_type='user' AND b.source_country=%s AND b.snapshot_week=%s AND u.account_number = ANY(%s) ORDER BY {p['lsort_col']} {direction} NULLS LAST, b.product_reference """, (p["country"], snapshot_week, acc_nos))

# group account -> user -> lines (line order preserved from SQL) by_acc: dict = {} for ln in lines: users = by_acc.setdefault(ln["account_number"], {}) u = users.setdefault(ln["ecom_user_id"], {"user_email": ln["user_email"], "user_name": ln["user_name"], "lines": []}) u["lines"].append(ln)

accounts = [] for r in acc_rows: # preserve account page order an = r["account_number"] m = meta.get(an, {}) users = [{"ecom_user_id": uid, uv} for uid, uv in by_acc.get(an, {}).items()] users.sort(key=lambda uu: -sum((l["orders_count"] or 0) for l in uu["lines"])) accounts.append({"account_number": an, "name": m.get("name"), "cluster_name": m.get("cluster_name"), "tot_orders": r["tot_orders"], "tot_units": r["tot_units"], "n_users": r["n_users"], "users": users})

return templates.TemplateResponse(request, "_buying_profile_explorer.html", {base, "accounts": accounts, "has_next": has_next}) `

  • [ ] Step 2: Parse-check app.py

Run: venv\Scripts\python.exe -c "import ast; ast.parse(open(r'dashboard/app.py',encoding='utf-8').read()); print('app.py OK')" Expected: app.py OK.

  • [ ] Step 3: Commit

`bash git add dashboard/app.py git commit -m "feat(explorer): buying-profile-explorer route (paged accounts + grouped lines)" `

Task 3: Template + nav

Files:

  • Create: dashboard/templates/_buying_profile_explorer.html
  • Modify: dashboard/templates/home.html

  • [ ] Step 1: Create the template

Create dashboard/templates/_buying_profile_explorer.html:

`html

{% macro plink(label, key, val) -%} {{ label }} {%- endmacro %}

{% if empty %}

No buying-profile snapshot computed yet.
{% elif no_match %}
No account found for "{{ p.q }}" in {{ p.country }}.
{% elif not accounts %}
No accounts on this page.
{% else %}
Snapshot week of {{ snapshot_week }} · totals are over each user's stored Top-10 lines (a ranking proxy, not full account revenue).
{% for a in accounts %}
{{ a.account_number }} — {{ a.name or '(no name)' }} {% if a.cluster_name %}[{{ a.cluster_name }}] {% endif %} · {{ a.n_users }} user(s) · {{ a.tot_orders }} orders · {{ a.tot_units | round(0) | int }} units
{% for u in a.users %}
{{ u.user_name or u.user_email or 'user' }} {{ u.ecom_user_id }}
{% for l in u.lines %} {% endfor %}
#Product {{ plink('Orders','lsort','orders') }} {{ plink('Units','lsort','units') }}
{{ l.rank_freq }} {{ (l.product_description or l.product_reference)[:54] }} · {{ l.product_reference }} {{ l.orders_count }} {{ l.units_qty | round(0) | int }}
{% endfor %}
{% endfor %}

{% if not p.q %}

page {{ page + 1 }}
{% endif %} {% endif %} `

  • [ ] Step 2: Register the panel in home.html

In dashboard/templates/home.html, find the Buying Profile panel block inside the Customers section (search for id="buying-profile"). Immediately AFTER its closing (the one closing

), insert:

`html

Buying Profile Explorer — all accounts, grouped by user & product (paginated)

` (If the exact id="buying-profile" panel block boundaries are unclear, place this new panel anywhere within the
block, as a sibling
.)

  • [ ] Step 3: Parse-check + commit

Run: venv\Scripts\python.exe -c "print('templates are not python — skipping parse')" (templates are Jinja; no Python parse). Confirm the two files were saved.

`bash git add dashboard/templates/_buying_profile_explorer.html dashboard/templates/home.html git commit -m "feat(explorer): grouped table template + Customers-tab nav" `

  • [ ] Step 4: Live verify (controller/Pierre — service restart is gated)

Restart: nssm restart LeadContagionDashboard. Then in the dashboard → Customers → Buying Profile Explorer: confirm accounts list biggest-first, each shows users → product lines, the account filter narrows to one account, the Orders/Units column links re-sort lines, and Prev/Next paginate.

---

Self-review

  • Spec coverage: grain account→user→line → Task 2 grouping + Task 3 template; paginate-by-account 20/page → Task 2 (LIMIT PAGE_SIZE+1 OFFSET) + Task 3 pager; biggest-buyers-first default → asort default orders desc; account filter → q branch; sortable line/account columns → asort/lsort whitelist + column-header links; own panel → Task 3 nav; injection guard → Task 1 parser (whitelisted asort_col/lsort_col); totals caveat → template hint line.
  • Injection: only whitelisted identifiers (tot_orders/tot_units/n_users/rank_freq/orders_count/units_qty) and ASC/DESC are interpolated into SQL; all other values parameterized. Task 1 test asserts the resolved columns are bare identifiers.
  • Single-DB: every query hits Postgres only (buying_profile_weekly, ecom_users, ecom_accounts, account_firmographic_community, firmographic_community_labels, ecom_products). No cross-DB join.
  • Type consistency: parser returns page/asort/lsort/dir/country/q/asort_col/lsort_col; the route reads exactly those keys; template reads p.q/p.asort/p.lsort/p.dir/p.country and accounts[].users[].lines[] fields produced by the route.
  • has_next: computed by fetching PAGE_SIZE+1 and trimming — correct end-of-list detection without a COUNT.

Notes for the implementer

  • buying_profile_weekly stores each user's Top-N lines only, so account totals here are a ranking proxy (stated in the UI). That's intentional — do not try to recompute full totals from ecom_order_lines (cross-DB).
  • The account filter q path returns ALL matching accounts (usually one) with no pager — that's intended.
  • The dir flip in the plink macro lets a header click toggle asc/desc on the line columns; account sort uses the dir select.