| """routes_grid.py β X2's write seam over HTTP: the SECOND adapter on `core.grid_events` (EXIT-1c). |
| |
| `POST /api/v1/grid/events` takes the component's event objects VERBATIM β the same objects the |
| Streamlit host receives through the component value slot, unchanged β and runs them through the |
| same `core.grid_events.handle_events` the Streamlit adapter runs. That is the whole point of |
| EXIT-1a: one implementation of every permission wall, two transports. A route that re-validated |
| anything here would be a second wall to keep in step, and the two would drift on the first change. |
| |
| DEDUP IS PER REQUEST, and that is a deliberate limit, not an oversight. The Streamlit adapter's |
| `seen_ids` lives in `st.session_state` β genuinely per user session β because the client resends |
| its recent 24-event window on every emit inside one page session. A stateless API has no such |
| dict, and inventing a per-user server-side one would be exactly the resident per-tenant state |
| EXIT-4a exists to remove. So: ids are deduped WITHIN a request body (the resend window's whole |
| purpose β a batch that repeats an id processes it once), and a genuinely replayed request is |
| handled by the operations being idempotent. The one event where a replay is observable is |
| `add_to_list` (it would add the same pids to the same cohort twice β a set union, so the |
| membership is unchanged, but the toast count repeats). Noted rather than papered over; a |
| server-side idempotency key belongs with D2's session mirror (C-2). |
| |
| STORE DOWN = 503. `fallback_ws` is None on this adapter, so `core.grid_events` raises |
| `StoreUnavailable` rather than writing to an in-memory workspace no API request could ever read |
| back. A 200 over a write that evaporated is the failure this rule exists to prevent. |
| """ |
| import datetime as dt |
| import time |
|
|
| from fastapi import APIRouter, Body, Depends |
|
|
| from deps import Session, err, module_gate, perms, require_session |
|
|
| router = APIRouter(prefix="/api/v1") |
|
|
| MODULE = "customer_data" |
|
|
| |
| |
| _MAX_EVENTS = 24 |
|
|
|
|
| def _ctx(session: Session, fields, pids, **kw): |
| from core import grid_events |
| import aios_grid |
| import core.perm_scope as perm_scope |
|
|
| |
| |
| |
| |
| |
| scope_key = str(kw.get("scope_key") or "") |
| if scope_key == "product": |
| import modules.product_data as pd |
| from routes_products import MODULE as _PMOD, pd_fields |
|
|
| module, canonical = _PMOD, pd_fields(consolidated=True) |
| kw.setdefault("table", pd.TABLE_OPS) |
| hidden = perm_scope.hidden_keys(session.user, module, canonical) |
| elif scope_key.startswith("ut_"): |
| |
| |
| |
| import core.table_store as table_store |
|
|
| kw.setdefault("table", |
| table_store.make(f"{scope_key}_table_workspace", st=session.runtime)) |
| hidden = frozenset() |
| else: |
| module, canonical = MODULE, aios_grid.FIELDS |
| hidden = perm_scope.hidden_keys(session.user, module, canonical) |
| return grid_events.EventCtx( |
| uname=session.uname, allowed_pids=pids, fields=fields, admin=session.admin, |
| |
| |
| |
| hidden_keys=hidden, |
| |
| |
| |
| |
| |
| |
| |
| st=session.runtime, |
| fallback_ws=None, seen_ids={}, **kw) |
|
|
|
|
| |
| |
| |
| |
| _SCOPES = ("customer", "cohort", "product") |
|
|
|
|
| def _scope_or_400(raw): |
| """β REFUSE AN UNKNOWN SCOPE, never default it. A typo silently served as `customer` would |
| hand a user the whole book on a page they opened to see one cohort β the widening direction, |
| which is the one that must fail closed. |
| |
| Wave 18 (C3-UT): a `ut_`-prefixed scope names a USER TABLE and passes through here β |
| existence and the per-table wall are enforced by `routes_tables.ut_assembly` (404/403), |
| which every consumer of such a scope goes through. Passing an unknown ut key therefore |
| still fails closed, just one layer down where the store can actually be consulted.""" |
| scope = (raw or "customer").strip().lower() |
| if scope.startswith("ut_"): |
| return scope |
| if scope not in _SCOPES: |
| raise err(400, "bad_scope", |
| f"scope must be one of {', '.join(_SCOPES)} β refusing to guess") |
| return scope |
|
|
|
|
| @router.get("/workspace") |
| def workspace(scope: str = "customer", |
| session: Session = Depends(require_session)): |
| """The durable table workspace for this session: views, fields, overlays, folders. |
| |
| β WAVE 21 (C4): the wall is TOPIC-SHAPED, so the DEPENDENCY is session-only and each scope |
| asserts its own gate below. The old `module_gate("customer_data")` dependency 403'd a |
| `ut_*` workspace for any tenant whose catalogue omits the customer module (loopable/ |
| nurilab/gtmlab ship modules w/o it) β and the client then fell back to the shared |
| localStorage bucket + demo data, which is exactly the "new database shows RI's customer |
| fields" defect. A user table's real wall is `user_tables.may_open`, enforced inside |
| `ut_assembly` (404/403, one layer down where the store can be consulted). |
| |
| β WHY THIS EXISTS (S1βS2, 2026-07-30 β an X2 AMENDMENT, see the split doc). X2 fixes |
| `/customers` at the exact shape `verify_fields_contract.py` referees, so the workspace cannot |
| ride along in it without forking the thing that gate keeps single-sourced. Without a |
| workspace route the standalone shell's write path would be WRITE-ONLY: events persist |
| server-side and nothing reads them back on reload, so a saved view looks lost to the user |
| even though the store has it. This route closes that read-back gap at its own URL. |
| |
| `allowed_pids` is passed so a SHARED view's `memberPids` are re-scoped to THIS reader β the |
| wave-9 leak rule. Omitting it would hand a Fisch-scoped user a member list built by a |
| full-access user. |
| """ |
| from core import grid_events |
| from routes_customers import grid_assembly |
|
|
| scope = _scope_or_400(scope) |
| |
| |
| |
| if not (scope == "product" or scope.startswith("ut_")): |
| session.require(MODULE) |
| |
| |
| |
| if scope.startswith("ut_"): |
| from routes_tables import ut_assembly |
|
|
| storage_key = f"{session.tenant}:{scope}:{session.uname}" |
| try: |
| |
| |
| |
| |
| |
| |
| g = ut_assembly(session, scope, storage_key=storage_key, with_rows=False) |
| except grid_events.StoreUnavailable: |
| raise err(503, "store_unavailable", "the tenant store is unavailable") |
| workspace = g["workspace"] |
| workspace["overlays"] = g["ws"].get("overlays") or {} |
| |
| |
| |
| workspace["fields"] = g["fields"] |
| workspace["measures"] = g["measures"] |
| workspace["measureSets"] = g["measure_sets"] |
| |
| |
| |
| |
| |
| workspace["derived"] = {str(pid): cells for pid, cells in (g["derived"] or {}).items()} |
| workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)} |
| try: |
| from core import users as _users |
| workspace["userOptions"] = _users.assignable_people(tenant=session.tenant) |
| workspace["userAvatars"] = {k: v for k, v in |
| _users.avatar_map(tenant=session.tenant).items() |
| if str(v).startswith("data:image/")} |
| except Exception: |
| workspace["userOptions"] = [] |
| workspace["userAvatars"] = {} |
| workspace["scopeKey"] = scope |
| |
| |
| |
| |
| |
| workspace["limits"] = g.get("limits") or [] |
| return {"workspace": workspace} |
|
|
| |
| |
| |
| |
| |
| if scope == "product": |
| from routes_products import MODULE as PRODUCT_MODULE, product_assembly |
|
|
| |
| |
| session.require(PRODUCT_MODULE) |
| storage_key = f"{session.tenant}:product-list:{session.uname}:all:all" |
| try: |
| g = product_assembly(session, scope=scope, storage_key=storage_key) |
| except grid_events.StoreUnavailable: |
| raise err(503, "store_unavailable", "the tenant store is unavailable") |
| workspace = g["workspace"] |
| workspace["overlays"] = g["ws"].get("overlays") or {} |
| |
| |
| workspace["fields"] = g["fields"] |
| |
| |
| workspace["measures"] = g["measures"] |
| workspace["measureSets"] = g["measure_sets"] |
| |
| |
| |
| |
| workspace["derived"] = {str(pid): cells for pid, cells in (g["derived"] or {}).items()} |
| workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)} |
| try: |
| from core import users as _users |
| workspace["userOptions"] = _users.assignable_people(tenant=session.tenant) |
| workspace["userAvatars"] = {k: v for k, v in |
| _users.avatar_map(tenant=session.tenant).items() |
| if str(v).startswith("data:image/")} |
| except Exception: |
| workspace["userOptions"] = [] |
| workspace["userAvatars"] = {} |
| workspace["scopeKey"] = scope |
| return {"workspace": workspace} |
|
|
| |
| |
| |
| |
| |
| |
| |
| import core.perm_scope as perm_scope |
| team_id, agent = perm_scope.derive_pool_scope(session.user, MODULE) |
| bu = team_id if team_id is not None else "all" |
| if scope == "cohort": |
| storage_key = f"{session.tenant}:cohort:{session.uname}:{bu}" |
| else: |
| storage_key = f"{session.tenant}:customer-list:{session.uname}:{bu}:{agent or 'all'}" |
|
|
| |
| |
| |
| |
| |
| try: |
| g = grid_assembly(session, scope=scope, storage_key=storage_key) |
| except grid_events.StoreUnavailable: |
| raise err(503, "store_unavailable", "the tenant store is unavailable") |
| workspace = g["workspace"] |
| |
| |
| |
| |
| workspace["overlays"] = g["ws"].get("overlays") or {} |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| workspace["fields"] = g["fields"] |
| |
| |
| |
| |
| |
| workspace["measures"] = g["measures"] |
| workspace["measureSets"] = g["measure_sets"] |
| |
| |
| workspace["derived"] = {str(pid): cells for pid, cells in g["derived"].items()} |
| |
| |
| |
| workspace["viewer"] = {"name": session.uname, "isAdmin": bool(session.admin)} |
| try: |
| from core import users as _users |
| workspace["userOptions"] = _users.assignable_people(tenant=session.tenant) |
| |
| |
| |
| workspace["userAvatars"] = {k: v for k, v in |
| _users.avatar_map(tenant=session.tenant).items() |
| if str(v).startswith("data:image/")} |
| except Exception: |
| workspace["userOptions"] = [] |
| workspace["userAvatars"] = {} |
|
|
| |
| |
| |
| workspace["scopeKey"] = scope |
| if scope == "cohort": |
| workspace["cohortMode"] = True |
| |
| workspace["scopeChoice"] = True |
| return {"workspace": workspace} |
|
|
|
|
| |
| _TS_BUCKETS = ("week", "month", "quarter", "year") |
| _TS_MAX_BUCKETS = 120 |
| _TS_MAX_FIELDS = 12 |
| _TS_MAX_PIDS = 5000 |
| _TS_MAX_LAST_N = 120 |
| |
| |
| |
| |
| _TS_MAX_CELLS = 720 |
|
|
|
|
| def _ts_start_of(bucket, d): |
| """The calendar START of the bucket holding `d` (week = Monday, the vocabulary rule).""" |
| if bucket == "week": |
| return d - dt.timedelta(days=d.weekday()) |
| if bucket == "month": |
| return d.replace(day=1) |
| if bucket == "quarter": |
| return d.replace(month=((d.month - 1) // 3) * 3 + 1, day=1) |
| return d.replace(month=1, day=1) |
|
|
|
|
| def _ts_next(bucket, d): |
| if bucket == "week": |
| return d + dt.timedelta(days=7) |
| if bucket == "year": |
| return d.replace(year=d.year + 1) |
| step = 1 if bucket == "month" else 3 |
| m = d.month + step |
| return dt.date(d.year + (m - 1) // 12, (m - 1) % 12 + 1, 1) |
|
|
|
|
| def _ts_prev(bucket, d): |
| if bucket == "week": |
| return d - dt.timedelta(days=7) |
| if bucket == "year": |
| return d.replace(year=d.year - 1) |
| step = 1 if bucket == "month" else 3 |
| y, m = d.year, d.month - step |
| while m < 1: |
| m += 12 |
| y -= 1 |
| return dt.date(y, m, 1) |
|
|
|
|
| def _ts_label(bucket, start): |
| if bucket == "week": |
| return f"Wk of {start.strftime('%b %d')}" |
| if bucket == "month": |
| return start.strftime("%b %Y") |
| if bucket == "quarter": |
| return f"Q{(start.month - 1) // 3 + 1} {start.year}" |
| return str(start.year) |
|
|
|
|
| @router.post("/grid/timeseries") |
| def grid_timeseries(body: dict = Body(default=None), scope: str = "customer", |
| session: Session = Depends(module_gate(MODULE))): |
| """Pooled measure values per calendar bucket β the time-series view's data channel. |
| |
| C-TSWIN (wave 14, ruling R1): AS-OF semantics. Each metric's OWN stored window is |
| re-resolved per bucket with `today := min(bucket end, real today)` β `ytd` is cumulative |
| from Jan 1, `last_90_days` trailing, `all_time` cumulative ever; bucket N equals what the |
| grid's measure column would show if today were bucket N's end. Fixed-range (`custom`) |
| windows cannot slide and are dropped per field as `window_fixed`. Rows carry |
| `window: {kind, label}` so a cumulative row cannot be misread as periodic, and there is |
| NO `total` column β sliding windows overlap, so a sum of columns would double-count. |
| |
| The client sends the pids its view currently matches (the filter IS the scope); the server |
| intersects them with the session's book, so the request can only ever NARROW. Values are |
| POOLED aggregates computed by the semantic layer's own expression (the additivity law: an |
| average is computed at pool grain, never averaged over per-customer answers). A bucket |
| with no rows is 0 for a sum/count and null for anything else; a bucket that has not |
| STARTED yet is null for every kind β an unstarted period's YTD is unanswered, not 0. |
| """ |
| from core import measure_resolve |
| from harness import windows as _wn |
| from routes_customers import _pool_stamp, _team_agent, allowed_pids |
|
|
| |
| |
| |
| |
| |
| _sc = _scope_or_400(scope) |
| if _sc == "product" or _sc.startswith("ut_"): |
| raise err(400, "bad_timeseries", |
| "this surface has no time-series channel β measures are customer-grain") |
|
|
| body = body or {} |
| bucket = body.get("bucket") |
| if bucket not in _TS_BUCKETS: |
| raise err(400, "bad_timeseries", "bucket must be one of week, month, quarter, year") |
| raw_fields = body.get("fields") |
| if not isinstance(raw_fields, list) or not raw_fields: |
| raise err(400, "bad_timeseries", "fields must be a non-empty list of field keys") |
| if len(raw_fields) > _TS_MAX_FIELDS: |
| raise err(400, "bad_timeseries", f"at most {_TS_MAX_FIELDS} fields per request") |
| raw_pids = body.get("pids") |
| if not isinstance(raw_pids, list) or not raw_pids: |
| raise err(400, "bad_timeseries", "pids must be a non-empty list") |
| if len(raw_pids) > _TS_MAX_PIDS: |
| raise err(400, "bad_timeseries", f"at most {_TS_MAX_PIDS} pids per request") |
| try: |
| wanted = {int(p) for p in raw_pids} |
| except (TypeError, ValueError): |
| raise err(400, "bad_timeseries", "pids must be integers") |
| pool = wanted & {int(p) for p in allowed_pids(session)} |
| if not pool: |
| |
| |
| raise err(403, "out_of_scope", "none of those customers are in your book") |
|
|
| span = body.get("span") |
| if not isinstance(span, dict): |
| raise err(400, "bad_timeseries", "span must be {'lastN': n} or {'from': .., 'to': ..}") |
| today = time.strftime("%Y-%m-%d") |
| t = dt.date.fromisoformat(today) |
| n = span.get("lastN") |
| starts = [] |
| if n is not None: |
| if not isinstance(n, int) or isinstance(n, bool) or not (1 <= n <= _TS_MAX_LAST_N): |
| raise err(400, "bad_timeseries", f"lastN must be 1..{_TS_MAX_LAST_N}") |
| cur = _ts_start_of(bucket, t) |
| starts = [cur] |
| for _ in range(n - 1): |
| cur = _ts_prev(bucket, cur) |
| starts.append(cur) |
| starts.reverse() |
| else: |
| try: |
| d_from = dt.date.fromisoformat(str(span.get("from"))) |
| d_to = dt.date.fromisoformat(str(span.get("to"))) |
| except (TypeError, ValueError): |
| raise err(400, "bad_timeseries", "span.from/to must be ISO dates (YYYY-MM-DD)") |
| if d_from > d_to: |
| d_from, d_to = d_to, d_from |
| cur = _ts_start_of(bucket, d_from) |
| while cur <= d_to: |
| starts.append(cur) |
| if len(starts) > _TS_MAX_BUCKETS: |
| raise err(400, "bad_timeseries", |
| f"that span is more than {_TS_MAX_BUCKETS} {bucket} buckets - " |
| f"narrow it") |
| cur = _ts_next(bucket, cur) |
| if not starts: |
| raise err(400, "bad_timeseries", "the span holds no buckets") |
| ends = [_ts_next(bucket, s) - dt.timedelta(days=1) for s in starts] |
|
|
| rt = session.runtime |
| if not rt.available(): |
| raise err(503, "store_unavailable", "the tenant store is unavailable") |
| import modules.customer_data as cl_mod |
| |
| |
| ws = cl_mod.table_workspace(session.uname, consume_corrections=False) |
| fdefs = ws.get("fields") or {} |
| keys, dropped, seen = [], [], set() |
| for k in raw_fields: |
| k = str(k or "") |
| if not k or k in seen: |
| continue |
| seen.add(k) |
| fd = fdefs.get(k) |
| if not isinstance(fd, dict) or not isinstance(fd.get("measure"), dict): |
| |
| dropped.append({"field": k, "reason": "not_a_measure_field"}) |
| continue |
| keys.append(k) |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| mfields, slid_keys = [], [] |
| for k in keys: |
| m = dict(fdefs[k]["measure"]) |
| if (m.get("window") or {}).get("kind") == "custom": |
| |
| |
| dropped.append({"field": k, "reason": "window_fixed"}) |
| continue |
| mfields.append({"key": k, "measure": m}) |
| slid_keys.append(k) |
| |
| |
| if len(starts) * len(mfields) > _TS_MAX_CELLS: |
| raise err(400, "bad_timeseries", |
| "that ask is too wide - narrow the span or pick fewer metrics") |
|
|
| team_id, agent = _team_agent(session) |
| stamp = _pool_stamp(rt, team_id, agent) |
| problems = [] |
| bucket_bounds = [(s.isoformat(), e.isoformat()) for s, e in zip(starts, ends)] |
| |
| |
| answers = measure_resolve.series_values( |
| mfields, bucket, bucket_bounds, |
| team_id, frozenset(pool), today, stamp, rt.series_memo, |
| on_error=lambda tag, e: problems.append(str(e)[:200])) if mfields else {} |
|
|
| columns = [] |
| for s, e in zip(starts, ends): |
| col = {"key": s.isoformat(), "label": _ts_label(bucket, s), |
| "from": s.isoformat(), "to": e.isoformat()} |
| if e > t: |
| col["partial"] = True |
| columns.append(col) |
| rows = [] |
| for k in slid_keys: |
| ans = answers.get(k) |
| if ans is None: |
| dropped.append({"field": k, "reason": "unresolvable"}) |
| continue |
| vals_by = ans.get("values") or {} |
| agg_kind = str(ans.get("agg") or "sum") |
| zero_fill = agg_kind in ("sum", "count") |
| vals = [] |
| for s in starts: |
| if s > t: |
| |
| |
| vals.append(None) |
| else: |
| vals.append(vals_by.get(s.isoformat(), 0 if zero_fill else None)) |
| wspec = (fdefs[k].get("measure") or {}).get("window") |
| wnorm = _wn.normalize(wspec) or {} |
| rows.append({"field": k, "label": str(fdefs[k].get("label") or k)[:120], |
| "agg": agg_kind, "values": vals, |
| "window": {"kind": str(wnorm.get("kind") or ""), |
| "label": _wn.label(wspec)}}) |
| meta = {"pool": len(pool), "today": today, "bucket": bucket} |
| if dropped: |
| meta["dropped"] = dropped |
| if problems: |
| meta["problems"] = problems[:5] |
| return {"columns": columns, "rows": rows, "meta": meta} |
|
|
|
|
| |
| |
| |
| |
| _CAL_MAX_GROUPS = 31 |
| _CAL_MAX_FIELDS = 6 |
| _CAL_MAX_CELLS = 186 |
|
|
|
|
| @router.post("/grid/calendar_metrics") |
| def grid_calendar_metrics(body: dict = Body(default=None), scope: str = "customer", |
| session: Session = Depends(module_gate(MODULE))): |
| """Per-DAY measure values with each metric's own window slid to that day (C-CAL / R4). |
| |
| β WHY THIS EXISTS AT ALL. The calendar's summary cells used to aggregate the row VALUES the |
| grid already held β which for a measure column means "every member's YTD **as of today**", |
| summed and printed under a date in March. The number was arithmetically fine and semantically |
| a lie: it answered a question about today while sitting in a cell labelled with another day. |
| R4: a metric in a day cell is computed AS OF THAT DAY. |
| |
| The shape differs from the time-series channel in exactly one way, and it is the reason this |
| is a separate route rather than a parameter: **every group carries its OWN pid set**. A |
| calendar day holds the records the date field placed there, so day-to-day the subject |
| changes. One `allowed_pids` for the whole request β the TS channel's shape β would compute |
| each day over everybody, which is a different question again. |
| |
| Static (non-measure) fields are NOT served here. They have no window to slide, so the |
| client's own per-day aggregation over row values stays correct for them; sending them would |
| invite a second implementation of arithmetic that already works. |
| """ |
| from core import measure_resolve |
| from harness import windows as _wn |
| from routes_customers import _pool_stamp, _team_agent, allowed_pids |
|
|
| |
| |
| _sc_cal = _scope_or_400(scope) |
| if _sc_cal == "product" or _sc_cal.startswith("ut_"): |
| raise err(400, "bad_calendar_metrics", |
| "the product surface has no measure channel yet β measures are customer-grain") |
|
|
| body = body or {} |
| raw_groups = body.get("groups") |
| if not isinstance(raw_groups, list) or not raw_groups: |
| raise err(400, "bad_calendar_metrics", |
| "groups must be a non-empty list of {key: 'YYYY-MM-DD', pids: [...]}") |
| if len(raw_groups) > _CAL_MAX_GROUPS: |
| raise err(400, "bad_calendar_metrics", |
| f"at most {_CAL_MAX_GROUPS} days per request (one month)") |
| raw_fields = body.get("fields") |
| if not isinstance(raw_fields, list) or not raw_fields: |
| raise err(400, "bad_calendar_metrics", "fields must be a non-empty list of field keys") |
| if len(raw_fields) > _CAL_MAX_FIELDS: |
| raise err(400, "bad_calendar_metrics", f"at most {_CAL_MAX_FIELDS} metrics per request") |
| if len(raw_groups) * len(raw_fields) > _CAL_MAX_CELLS: |
| raise err(400, "bad_calendar_metrics", |
| "that ask is too wide - fewer days or fewer metrics") |
|
|
| book = {int(p) for p in allowed_pids(session)} |
| groups, seen_days = [], set() |
| for g in raw_groups: |
| if not isinstance(g, dict): |
| raise err(400, "bad_calendar_metrics", "every group must be an object") |
| day = str(g.get("key") or "") |
| try: |
| d = dt.date.fromisoformat(day) |
| except (TypeError, ValueError): |
| raise err(400, "bad_calendar_metrics", |
| "every group key must be an ISO date (YYYY-MM-DD)") |
| if day in seen_days: |
| raise err(400, "bad_calendar_metrics", f"day {day} appears twice") |
| seen_days.add(day) |
| raw_pids = g.get("pids") |
| if not isinstance(raw_pids, list): |
| raise err(400, "bad_calendar_metrics", "every group needs a pids list") |
| try: |
| wanted = {int(p) for p in raw_pids} |
| except (TypeError, ValueError): |
| raise err(400, "bad_calendar_metrics", "pids must be integers") |
| |
| |
| groups.append((day, d, frozenset(wanted & book))) |
| if sum(len(p) for _, _, p in groups) > _TS_MAX_PIDS: |
| raise err(400, "bad_calendar_metrics", |
| f"at most {_TS_MAX_PIDS} customer references per request") |
|
|
| rt = session.runtime |
| if not rt.available(): |
| raise err(503, "store_unavailable", "the tenant store is unavailable") |
| import modules.customer_data as cl_mod |
| ws = cl_mod.table_workspace(session.uname, consume_corrections=False) |
| fdefs = ws.get("fields") or {} |
| keys, dropped, seen = [], [], set() |
| for k in raw_fields: |
| k = str(k or "") |
| if not k or k in seen: |
| continue |
| seen.add(k) |
| fd = fdefs.get(k) |
| if not isinstance(fd, dict) or not isinstance(fd.get("measure"), dict): |
| |
| |
| dropped.append({"field": k, "reason": "not_a_measure_field"}) |
| continue |
| if ((fd["measure"].get("window") or {}).get("kind") == "custom"): |
| dropped.append({"field": k, "reason": "window_fixed"}) |
| continue |
| keys.append(k) |
| if not keys: |
| raise err(400, "bad_calendar_metrics", |
| "none of the requested fields are measure fields with a window that can " |
| "slide to a day") |
|
|
| today = time.strftime("%Y-%m-%d") |
| t = dt.date.fromisoformat(today) |
| team_id, agent = _team_agent(session) |
| stamp = _pool_stamp(rt, team_id, agent) |
| problems = [] |
| values = {k: {} for k in keys} |
| for day, d, pids in groups: |
| |
| |
| |
| if d > t or not pids: |
| for k in keys: |
| values[k][day] = None |
| continue |
| answers = measure_resolve.series_values( |
| [{"key": k, "measure": dict(fdefs[k]["measure"])} for k in keys], |
| "day", [(day, day)], team_id, pids, today, stamp, rt.series_memo, |
| on_error=lambda tag, e: problems.append(str(e)[:200])) |
| for k in keys: |
| ans = answers.get(k) |
| if ans is None: |
| values[k][day] = None |
| continue |
| vals_by = ans.get("values") or {} |
| zero_fill = str(ans.get("agg") or "sum") in ("sum", "count") |
| values[k][day] = vals_by.get(day, 0 if zero_fill else None) |
|
|
| for k in keys: |
| if all(v is None for v in values[k].values()): |
| |
| |
| dropped.append({"field": k, "reason": "unresolvable"}) |
| out = {"values": values, "today": today, |
| "windows": {k: {"kind": str((_wn.normalize( |
| (fdefs[k].get("measure") or {}).get("window")) or {}).get("kind") or ""), |
| "label": _wn.label((fdefs[k].get("measure") or {}).get("window"))} |
| for k in keys}} |
| if dropped: |
| out["dropped"] = dropped |
| if problems: |
| out["problems"] = problems[:5] |
| return out |
|
|
|
|
| @router.post("/grid/events") |
| def grid_events_route(body: dict = Body(default=None), |
| session: Session = Depends(require_session)): |
| """`{events: [<component event objects, verbatim>]}` β `{results, doc?, toast?}`. |
| |
| Wave 21 C4: session-only dependency, per-scope gate below β the write door must admit the |
| same sessions the read door (`/workspace`) admits, or a tenant without the customer module |
| can SEE its own user tables and not write to them.""" |
| from core import grid_events |
| from routes_customers import grid_assembly |
|
|
| events = (body or {}).get("events") |
| if events is None and isinstance(body, dict) and body.get("type"): |
| events = [body] |
| if not isinstance(events, list): |
| raise err(400, "bad_events", "expected {events: [...]}") |
| if len(events) > _MAX_EVENTS: |
| raise err(400, "too_many_events", |
| f"at most {_MAX_EVENTS} events per request (the client's resend window)") |
|
|
| |
| |
| |
| |
| scope = _scope_or_400((body or {}).get("scopeKey")) |
| |
| if not (scope == "product" or scope.startswith("ut_")): |
| session.require(MODULE) |
| |
| |
| |
| |
| |
| |
| |
| if scope == "product": |
| from routes_products import MODULE as PRODUCT_MODULE, product_assembly |
|
|
| session.require(PRODUCT_MODULE) |
| g = product_assembly(session, consume_corrections=False) |
| elif scope.startswith("ut_"): |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| from routes_tables import ut_write_ctx |
|
|
| g = ut_write_ctx(session, scope) |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| if g.get("limits"): |
| named = [e for e in events |
| if isinstance(e, dict) and (e.get("pid") is not None or e.get("pids"))] |
| if named: |
| lim = g["limits"][0] |
| raise err(409, "pid_scope_unresolved", |
| f"this database is served through the connector mirror and its rows " |
| f"cannot be listed in one window, so a change addressed to particular " |
| f"records ({len(named)} of {len(events)} here) cannot be admitted β " |
| f"{lim.get('cause') or 'the row set is unresolved'}. " |
| f"{lim.get('recommendation') or ''}".strip()) |
| else: |
| g = grid_assembly(session, scope=scope, consume_corrections=False) |
| |
| |
| |
| |
| |
| |
| ctx = _ctx(session, g["fields"], g["pids"], scope_key=scope, |
| measure_keys=frozenset(m["key"] for m in g["measures"]), |
| resolved_ids=frozenset(g["measure_sets"]), |
| cohort_ids=frozenset(c["id"] for c in g["lists"]), |
| measure_offer=tuple(g["measures"]), |
| visible_views=tuple(g["views"])) |
|
|
| |
| |
| results = [] |
| try: |
| for one in events: |
| eid = str(one.get("id") or "") if isinstance(one, dict) else "" |
| rerender = grid_events.handle_one(one, ctx) |
| results.append({"id": eid, "rerender": bool(rerender)}) |
| except grid_events.StoreUnavailable: |
| raise err(503, "store_unavailable", |
| "the tenant store is unavailable β none of your changes were saved") |
|
|
| out = {"results": results, "rerender": any(r["rerender"] for r in results)} |
| if ctx.out.doc is not None: |
| |
| |
| |
| |
| |
| |
| |
| out["doc"] = out["docPayload"] = ctx.out.doc |
| if ctx.out.toast is not None: |
| out["toast"] = ctx.out.toast |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| if out["rerender"] and scope in ("customer", "cohort") and any( |
| isinstance(e, dict) and e.get("type") == "field_upsert" |
| and str(((e.get("field") or {}) if isinstance(e.get("field"), dict) else {}) |
| .get("key") or "").startswith("measure_") |
| for e in events): |
| try: |
| fresh = grid_assembly(session, scope=scope, consume_corrections=False) |
| out["derived"] = {str(pid): cells for pid, cells in fresh["derived"].items()} |
| except Exception: |
| pass |
| |
| |
| |
| |
| |
| |
| return out |
|
|