#!/usr/bin/env python3
"""The database map as one page for a person: every table, column and key, the catalogue's
spine drawn, the owner's own words mapped to the tables.
Sources of truth: app/models/ (the SQLAlchemy metadata, read through `docmap.load_models`) and
`docmap.AREAS` (which tables a reader sees together). Nothing about a column is typed here: the
names, types, keys, constraints and docstrings are the live metadata every time it runs. What is
typed is the drawing's layout, the one-line role under each box, the words for the owner's
vocabulary and each area's sentence; those are the parts a person reads first, and a wrong one
is visible at a glance. The checked, durable form of the same map is the Relationships block
`docmap.py` writes into docs/DATA-MODEL.md; this page is what the owner opens.
python3 main/scripts/dbmap.py --html /path/to/out.html
Self-contained HTML (one file, two themes, no library): publish it as is.
"""
from __future__ import annotations
import argparse
import datetime as dt
import html
import pathlib
import subprocess
import sys
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
from docmap import MAIN, ROOT, areas_of, load_models # noqa: E402
# --- the owner's words -------------------------------------------------------------------------
WORDS = [
("Listings", "listings", "One product on one shop's shelf. The shop's own words are the listed_* "
"columns; a pin or an ignore on the row is the decided layer at this level."),
("Products", "products", "One size of one variation, under product_lines (the real "
"product, one page each) and brands. variation_aliases "
"is the vocabulary a wording resolves through."),
("Brands", "brands", "One row per house however the shops spell it; an alias points at its house through "
"canonical_id."),
("Collectors", "sources", "A collector's settings and its kill switch. Each pass is a row in "
"collection_runs. Which shops a collector reads is code "
"(services/collectors/registry.py), not a table."),
("Airports", "locations", "One airport shop (iata, visible) under "
"retailers, the operator. Opening hours are readings in "
"airport_hours."),
("Prices", "price_observations", "Append-only: one row per sighting, with the currency the page declared and the "
"run it came from. The fragment behind it is raw_records."),
("Medals", "awards", "A competition result on a product."),
("Your decisions", "overrides", "A field a person set over what the rules produced; also "
"listings.pinned_product_id and ignored_at, the merge queue "
"merge_candidates and the record product_merges."),
("People", "accounts", "The one identity table. Every *_by and *_id who-column elsewhere "
"points here."),
]
AREA_INTRO = {
"Catalogue": "What is sold and where: a retailer runs airport shops; a brand has lines, a line has products "
"(one size of one variation), and a listing is one product on one shop's shelf.",
"Collection": "Who reads the shops and what came back: a source is one collector's settings, a run is one pass, "
"and every run leaves the fragment it read and the price it saw.",
"Decisions and merges": "What a person decided over what the rules produced: an override on any field, the "
"merge queue and the record of every merge, and the queue for a fresh read.",
"Quality": "Whether the data is true: verification runs and their per-listing checks, and the audit's snapshots.",
"Accounts": "Who can sign in and what they may do. Every who-column elsewhere points at accounts.",
"Discussion": "Conversations on every subject, the asks, flags and acknowledgements on them, the inbox, and "
"mail's ledger.",
"Client and owner surfaces": "What the client and the owner record on the site: the feature ranking, the quote, "
"the to-do list with its hand-ins, and the running list's states.",
"Editorial": "Text people wrote: articles of three kinds, and the newsletter list.",
}
# --- the spine drawing -------------------------------------------------------------------------
BOX_W, BOX_H = 190, 58
NODES = { # table: (x, y, the one-line role under the name)
"sources": (40, 30, "a collector's settings"),
"collection_runs": (300, 30, "one pass of a collector"),
"retailers": (40, 170, "one operator"),
"locations": (300, 170, "an airport shop · visible"),
"brands": (40, 330, "alias → house: canonical_id"),
"product_lines": (300, 330, "the real product"),
"products": (560, 250, "one size of one variation"),
"listings": (820, 170, "one product on one shelf"),
"awards": (820, 390, "a medal on a product"),
"raw_records": (1080, 100, "the fragment read, per run"),
"price_observations": (1080, 240, "a price at a moment"),
}
EMPHASIS = {"products", "listings"}
# (points, label, label position, style): an arrow leaves the table that holds the key.
EDGES = [
([(300, 59), (230, 59)], "source_id", (265, 45), "key"),
([(300, 199), (230, 199)], "retailer_id", (265, 185), "key"),
([(300, 359), (230, 359)], "brand_id", (265, 345), "key"),
([(560, 279), (525, 279), (525, 359), (490, 359)], "line_id", (525, 319), "key"),
([(655, 308), (655, 420), (135, 420), (135, 388)], "brand_id", (395, 420), "key"),
([(820, 199), (785, 199), (785, 279), (750, 279)], "product_id", (785, 239), "key"),
([(860, 170), (860, 140), (395, 140), (395, 170)], "location_id", (627, 140), "key"),
([(860, 390), (860, 360), (700, 360), (700, 308)], "product_id", (780, 360), "key"),
([(1080, 129), (1045, 129), (1045, 187), (1010, 187)], "listing_id", (1045, 158), "key"),
([(1080, 269), (1045, 269), (1045, 211), (1010, 211)], "listing_id", (1045, 240), "key"),
([(1270, 269), (1300, 269), (1300, 14), (395, 14), (395, 30)], "run_id", (800, 14), "key"),
([(1175, 100), (1175, 14)], "", None, "join"),
([(135, 88), (135, 110), (265, 110), (265, 178), (300, 178)], "paired in code", (200, 110), "code"),
]
VIEW_W, VIEW_H = 1330, 470
def svg_spine() -> str:
out = [f'")
return "\n".join(out)
# --- the cards -----------------------------------------------------------------------------------
def human_type(t: str) -> str:
t = str(t).upper()
if t.startswith("VARCHAR("):
return f"text ≤{t[8:-1]}"
if t.startswith("NUMERIC("):
return f"number ({t[8:-1].replace(' ', '')})"
return {"TEXT": "text", "INTEGER": "integer", "BOOLEAN": "yes / no", "DATETIME": "timestamp",
"JSON": "json", "JSONB": "json"}.get(t, t.lower())
def first_paragraph(doc: str) -> str:
para = doc.split("\n\n", 1)[0] if doc else ""
return " ".join(para.split())
def card(t, doc: str, module: str, pointed: dict[str, list[tuple[str, str]]]) -> str:
cols = list(t.columns)
single_idx = {next(iter(i.columns)).name for i in t.indexes if len(i.columns) == 1}
multi_idx = [i for i in t.indexes if len(i.columns) > 1]
uniques = [k for k in t.constraints if k.__class__.__name__ == "UniqueConstraint"]
single_uq = {next(iter(k.columns)).name for k in uniques if len(k.columns) == 1}
multi_uq = [k for k in uniques if len(k.columns) > 1]
rows = []
for c in cols:
fk = next(iter(c.foreign_keys)).column if c.foreign_keys else None
marks = []
if c.primary_key:
marks.append('PK')
if fk is not None:
marks.append('FK')
if c.unique or c.name in single_uq:
marks.append('U')
if c.index or c.name in single_idx:
marks.append('I')
to = ""
if fk is not None:
same = fk.table.name == t.name
to = f'{fk.table.name}.{fk.name}'
if same:
to += ' (itself)'
cls = " opt" if (c.nullable and not c.primary_key) else ""
rows.append(f'
{c.name}'
f'{"?" if cls else ""}
'
f'
{html.escape(human_type(c.type))}
{" ".join(marks)}
'
f'
{to}
')
facts = []
if multi_uq:
facts.append("Unique together: " + "; ".join(
"(" + ", ".join(f"{c.name}" for c in k.columns) + ")" for k in multi_uq))
if multi_idx:
facts.append("Indexed together: " + "; ".join(
"(" + ", ".join(f"{c.name}" for c in i.columns) + ")" for i in multi_idx))
if pointed.get(t.name):
facts.append("Pointed at by: " + ", ".join(
f'{tb}.{cn}' for tb, cn in pointed[t.name] if tb != t.name)
or "itself")
para = first_paragraph(doc)
doc_html = f'
{html.escape(para)}
' if para else '
No docstring on the model yet.
'
colnames = " ".join(c.name.lower() for c in cols)
return (f''
f'
{t.name}
'
f'{len(cols)} columns · models/{module}.py'
f'{doc_html}'
f'
'
for w, t, p in WORDS)
sections = []
for area, names in areas:
intro = AREA_INTRO.get(area, "")
sections.append(
f'
'
f'
{html.escape(area)}
{f"
{html.escape(intro)}
" if intro else ""}
'
+ "".join(card(tables[n], docs.get(n, ""), modules.get(n, "?"), pointed) for n in names)
+ "
")
return f"""Duty Free Professor Database Map
Duty Free Professor · internal
The database, table by table
Every table the app has, what each column holds, which columns are keys and where they point. Read from the models, not from memory, so what is here is what Postgres has.
{len(tables)} tables
{n_cols} columns
{n_fk} keys between them
from main/app/models/ at {html.escape(sha)}, {today}
PK primary keyFK points at another tableU uniqueI indexedname? may be empty
In your words
The things you name, and the tables that hold them.
{words}
{svg_spine()}
The catalogue's spine. An arrow leaves the table that holds the key and names its column; the dotted line is a pairing that lives in code, not in a key. The decided layer (overrides, merges, a pin or an ignore on a listing) and the who-columns that point at accounts are in the cards below. Click a box to jump to its table.
{"".join(sections)}
"""
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--html", required=True, help="where to write the page")
a = ap.parse_args()
out = pathlib.Path(a.html)
out.write_text(build())
print(f"dbmap: wrote {out} ({out.stat().st_size // 1024} KB)")
return 0
if __name__ == "__main__":
sys.exit(main())