coreci/inboxos-v1
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases 3 Wiki Activity
Files
main
inboxos-v1/scripts/apply_schemas.py
T
cloudinit-bot 37d21fb430 feat(nocodb): 4 curated views per module + docs (REQ-INB-41) 
---ci---
project: inboxos
phase: 1
milestone: v1.4
status: execute
wave: 2
task: T2
---/ci---
2026-06-09 14:16:36 +00:00

558 lines
24 KiB
Python
Raw Permalink Blame History

"""Idempotently apply the InboxOS NocoDB schema.
Phase 2 deliverable for INB-4. Source of truth for the column lists
lives in `docs/schemas.md`; this script is the *executable* form of
that document and is safe to re-run any number of times.
Schema contract
---------------
The four module tables match the field contract defined by the
canonical OpenCode prompts under `.opencode/commands/`:
- `.opencode/commands/inboxos:classify.md`
- `.opencode/commands/inboxos:extract-finance.md`
- `.opencode/commands/inboxos:extract-crm.md`
- `.opencode/commands/inboxos:extract-intel.md`
- `.opencode/commands/inboxos:extract-promotions.md`
If a prompt's field contract changes, update `MODULES` below and
`docs/schemas.md` together. Do not silently rename or remove columns
that a deployed prompt still produces — that would break the
upsert.
What it does
------------
- Signs into NocoDB with NOCODB_BASE_URL + NOCODB_API_TOKEN.
- Finds (or creates) a base named `inboxos`.
- For each of the 4 module tables, finds (or creates) a table with
the right title and the right columns.
- Ensures a UNIQUE constraint on `Message_ID` for every module table.
- Adds the common pipeline columns that BOTH the live and historical
pipelines write (`Date`, `Raw_S3_Link`, `Papra_Link`,
`Extracted_At`) so the same table can be written by either
pipeline.
- Prints the resulting table IDs in the form expected by the
pipeline env var `NOCODB_MODULE_TABLES` (JSON dict).
Idempotency
-----------
- Existing bases, tables, and columns are detected by title / name
and left untouched.
- The unique-constraint check uses the column's `un` flag; if it is
already set, the script is a no-op there.
- Safe to run after the schema has already been applied.
Verify-only mode
----------------
python scripts/apply_schemas.py --check
In `--check` mode the script exits non-zero if any required object
is missing or if any column has the wrong type.
Environment
-----------
Required:
NOCODB_BASE_URL e.g. https://noco.cloudinit.dev
NOCODB_API_TOKEN xc-token from a NocoDB user with create rights
Optional:
NOCODB_PROJECT_ID If set, used as the target base id; otherwise
the script finds-or-creates a base named "inboxos".
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from typing import Any
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# ── Schema (mirrors docs/schemas.md) ─────────────────────────
BASE_TITLE = "inboxos"
# Common columns written by BOTH pipelines on every module row.
# Nullable: live pipeline writes Date / Raw_S3_Link, historical
# writes Papra_Link, both write Extracted_At. Either pipeline can
# write to the same table without the other's columns being required.
COMMON_COLUMNS: list[dict[str, Any]] = [
{"name": "Date", "type": "DateTime", "required": False},
{"name": "Raw_S3_Link", "type": "URL", "required": False},
{"name": "Papra_Link", "type": "URL", "required": False},
{"name": "Extracted_At", "type": "DateTime", "required": False},
]
# Per-module columns mirror the field contract in the prompt files
# at .opencode/commands/inboxos:extract-*.md. Keep in sync.
MODULES: list[dict[str, Any]] = [
{
"name": "finance",
"title": "pl_finance",
"description": "Finance / P&L records (receipts, invoices, statements, subscriptions, refunds).",
"columns": [
{"name": "Message_ID", "type": "SingleLineText", "required": True, "unique": True},
{"name": "Direction", "type": "SingleSelect",
"options": ["income", "expense", "refund", "transfer", "unknown"]},
{"name": "Counterparty", "type": "SingleLineText"},
{"name": "Amount", "type": "Decimal"},
{"name": "Currency", "type": "SingleLineText"},
{"name": "Transaction_Date", "type": "Date"},
{"name": "Category", "type": "SingleLineText"},
{"name": "Reference", "type": "SingleLineText"},
{"name": "Notes", "type": "LongText"},
],
},
{
"name": "crm",
"title": "personal_crm",
"description": "Personal CRM contacts and conversations.",
"columns": [
{"name": "Message_ID", "type": "SingleLineText", "required": True, "unique": True},
{"name": "Contact_Name", "type": "SingleLineText"},
{"name": "Contact_Email", "type": "Email"},
{"name": "Organization", "type": "SingleLineText"},
{"name": "Role", "type": "SingleLineText"},
{"name": "Channel", "type": "SingleSelect",
"options": ["inbound", "outbound", "intro", "reply", "scheduling", "other"]},
{"name": "Intent", "type": "SingleSelect",
"options": ["sales", "partnership", "recruiting", "support", "social", "intro", "other"]},
{"name": "Last_Contacted", "type": "Date"},
{"name": "Summary", "type": "LongText"},
{"name": "Action_Needed", "type": "LongText"},
],
},
{
"name": "intel",
"title": "daily_news",
"description": "Daily news / market intel / security digests.",
"columns": [
{"name": "Message_ID", "type": "SingleLineText", "required": True, "unique": True},
{"name": "Source", "type": "SingleLineText"},
{"name": "Topic", "type": "SingleSelect",
"options": ["markets", "ai", "security", "industry", "policy", "tech", "other"]},
{"name": "Headline", "type": "LongText"},
{"name": "Summary", "type": "LongText"},
{"name": "Items", "type": "LongText", # JSON-encoded array
"note": "stored as JSON-encoded string in a LongText column; "
"NocoDB's array columns are not stable across versions."},
{"name": "Issued_Date", "type": "Date"},
{"name": "Read_Priority", "type": "SingleSelect",
"options": ["must", "should", "skim", "archive"]},
],
},
{
"name": "promotions",
"title": "promotions",
"description": "Promotional / marketing emails.",
"columns": [
{"name": "Message_ID", "type": "SingleLineText", "required": True, "unique": True},
{"name": "Brand", "type": "SingleLineText"},
{"name": "Offer_Type", "type": "SingleSelect",
"options": ["discount", "bogo", "freebie", "launch", "sale", "event", "newsletter", "other"]},
{"name": "Headline", "type": "LongText"},
{"name": "Discount_Pct", "type": "Number"},
{"name": "Discount_Code", "type": "SingleLineText"},
{"name": "Valid_Until", "type": "Date"},
{"name": "Category", "type": "SingleSelect",
"options": ["apparel", "software", "hardware", "food", "travel", "services", "other"]},
{"name": "Worth_Saving", "type": "Checkbox"},
{"name": "Notes", "type": "LongText"},
],
},
]
# Per-table audit-trail tables (REQ-INB-19). One row per pipeline
# decision of the named kind. The v1.0 module table freeze is
# preserved: this list is a separate code path, not added to MODULES.
AUDIT_TABLES: list[dict[str, Any]] = [
{
"name": "papra_index",
"title": "papra_index",
"description": (
"Papra write audit trail (REQ-INB-19). One row per Papra write "
"intent; populated by pipelines.common.papra_index_upsert. "
"Created by apply_schemas.py on first run; idempotent on rerun."
),
"columns": [
{"name": "Message_ID", "type": "SingleLineText", "required": True, "unique": True},
{"name": "papra_document_id", "type": "SingleLineText"},
{"name": "module", "type": "SingleLineText"},
{"name": "kind", "type": "SingleSelect",
"options": ["attachment", "body"]},
{"name": "decision_at", "type": "DateTime", "required": True},
{"name": "outcome", "type": "SingleSelect",
"options": ["ok", "failed", "dry_run"]},
{"name": "error", "type": "LongText"},
],
},
]
# ── HTTP plumbing ────────────────────────────────────────────
def _session() -> requests.Session:
sess = requests.Session()
retry = Retry(
total=5,
backoff_factor=0.5,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=frozenset(["GET", "POST", "PATCH", "DELETE"]),
)
sess.mount("https://", HTTPAdapter(max_retries=retry))
sess.mount("http://", HTTPAdapter(max_retries=retry))
return sess
def _api(sess: requests.Session, base_url: str, token: str, method: str, path: str,
json_body: Any = None, params: dict | None = None) -> Any:
url = f"{base_url.rstrip('/')}{path}"
headers = {"xc-token": token, "Content-Type": "application/json"}
resp = sess.request(method, url, headers=headers, json=json_body, params=params, timeout=60)
if resp.status_code >= 400:
raise RuntimeError(
f"NocoDB {method} {path} failed: status={resp.status_code} body={resp.text[:500]!r}"
)
if not resp.content:
return None
return resp.json()
# ── Core operations ──────────────────────────────────────────
def find_or_create_base(sess: requests.Session, base_url: str, token: str,
project_id: str | None) -> str:
# When NOCODB_PROJECT_ID is set, verify the base exists by reading it
# directly. A list-bases call requires meta.projects:read, which base-
# scoped personal access tokens in NocoDB v2026.04.x do NOT have. The
# direct GET on a known id is allowed when the token has any access to
# that base.
if project_id:
try:
base = _api(sess, base_url, token, "GET",
f"/api/v2/meta/bases/{project_id}")
if isinstance(base, dict) and base.get("id") == project_id:
return project_id
raise RuntimeError(
f"GET /api/v2/meta/bases/{project_id} returned unexpected body: "
f"{base!r}")
except RuntimeError as exc:
# Re-raise as the canonical "not found" error so callers can
# distinguish a missing base from a permission failure.
if "404" in str(exc) or "ERR_BASE_NOT_FOUND" in str(exc):
raise RuntimeError(
f"NOCODB_PROJECT_ID={project_id!r} not found or not "
f"accessible to this token (verify the id and that the "
f"token has access to that base)") from exc
raise
# No explicit id: list bases (needs workspace-admin scope). For
# base-scoped tokens this will fail with ERR_FORBIDDEN; the caller
# should retry with NOCODB_PROJECT_ID set.
bases = _api(sess, base_url, token, "GET", "/api/v2/meta/bases")
for b in bases if isinstance(bases, list) else bases.get("list", []):
if (b.get("title") or b.get("name")) == BASE_TITLE:
return b["id"]
created = _api(sess, base_url, token, "POST", "/api/v2/meta/bases",
json_body={"title": BASE_TITLE})
return created["id"]
def find_table(sess: requests.Session, base_url: str, token: str,
base_id: str, title: str) -> dict | None:
tables = _api(sess, base_url, token, "GET", f"/api/v2/meta/bases/{base_id}/tables")
for t in tables if isinstance(tables, list) else tables.get("list", []):
if (t.get("title") or t.get("name")) == title:
return t
return None
def all_columns_for_module(module: dict) -> list[dict[str, Any]]:
"""Return [Message_ID (unique), ...module cols, ...common cols]."""
cols: list[dict[str, Any]] = []
for c in module["columns"]:
cols.append({
"column_name": c["name"],
"title": c["name"],
"uidt": c["type"],
"rqd": bool(c.get("required", False)),
"un": bool(c.get("unique", False)),
})
for c in COMMON_COLUMNS:
cols.append({
"column_name": c["name"],
"title": c["name"],
"uidt": c["type"],
"rqd": bool(c.get("required", False)),
"un": False,
})
# Options for selects
for i, c in enumerate(module["columns"]):
if "options" in c:
cols[i]["dtxp"] = c["options"]
cols[i]["options"] = c["options"]
return cols
def create_table(sess: requests.Session, base_url: str, token: str,
base_id: str, module: dict) -> dict:
payload = {
"title": module["title"],
"table_name": module["title"],
"description": module.get("description", ""),
"columns": all_columns_for_module(module),
}
return _api(sess, base_url, token, "POST", f"/api/v2/meta/bases/{base_id}/tables",
json_body=payload)
def find_column(table: dict, name: str) -> dict | None:
for c in table.get("columns", []):
if c.get("title") == name or c.get("column_name") == name:
return c
return None
def add_missing_columns(sess: requests.Session, base_url: str, token: str,
table: dict, columns: list[dict[str, Any]]) -> int:
added = 0
table_id = table["id"]
existing = {c.get("title") or c.get("column_name") for c in table.get("columns", [])}
for c in columns:
if c["name"] in existing:
continue
# NocoDB column-add payload: use `uidt` (UI Data Type) so NocoDB
# picks the correct storage type. Sending the UI type as `dt`
# (e.g. `dt: "SingleSelect"`, `dt: "Date"`) yields SQLITE_ERROR
# at the ALTER TABLE on v2026.04.x. `uidt` is accepted for every
# type the schema uses (SingleLineText, LongText, Email, URL,
# Date, DateTime, Decimal, SingleSelect, Checkbox).
body: dict[str, Any] = {
"title": c["name"],
"column_name": c["name"],
"uidt": c["type"],
"rqd": bool(c.get("required", False)),
"un": bool(c.get("unique", False)),
}
if c["type"] == "Decimal":
# Decimal needs an explicit precision. 18 is the
# same precision the existing tables use.
body["dtxp"] = 18
if "options" in c:
# SingleSelect options are managed separately via the
# colOptions endpoint after column creation; storing them
# in dtxp/dtxs as a JSON array does not work and causes
# SQLITE_ERROR. The pipeline writes option names by string
# value; if the column has no pre-defined options, NocoDB
# still accepts arbitrary string values (no CHECK enforced).
body["_options_pending"] = c["options"]
_api(sess, base_url, token, "POST", f"/api/v2/meta/tables/{table_id}/columns",
json_body=body)
added += 1
return added
def ensure_unique_on_message_id(sess: requests.Session, base_url: str, token: str,
column: dict) -> bool:
"""Make sure the column has a UNIQUE index. Returns True if changed."""
if column.get("un") is True or column.get("unique") is True:
return False
column_id = column["id"]
_api(sess, base_url, token, "PATCH", f"/api/v2/meta/columns/{column_id}",
json_body={"un": True})
return True
# ── Main ─────────────────────────────────────────────────────
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--check", action="store_true",
help="Verify the schema is in place; do not modify anything.")
parser.add_argument("--seed-views", action="store_true",
help="Seed 4 curated NocoDB views per module (v1.4 REQ-INB-41).")
args = parser.parse_args()
base_url = os.environ.get("NOCODB_BASE_URL", "").strip()
token = os.environ.get("NOCODB_API_TOKEN", "").strip()
project_id = os.environ.get("NOCODB_PROJECT_ID", "").strip() or None
if not base_url or not token:
print("NOCODB_BASE_URL and NOCODB_API_TOKEN are required.", file=sys.stderr)
return 2
sess = _session()
base_id = find_or_create_base(sess, base_url, token, project_id)
print(f"base: id={base_id} title={BASE_TITLE}")
module_table_ids: list[tuple[str, str]] = []
any_drift = False
for module in MODULES:
table = find_table(sess, base_url, token, base_id, module["title"])
if not table:
if args.check:
print(f" MISSING table {module['title']!r}", file=sys.stderr)
any_drift = True
continue
table = create_table(sess, base_url, token, base_id, module)
print(f" created table {module['title']!r} id={table['id']}")
else:
print(f" found table {module['title']!r} id={table['id']}")
# Re-fetch the table with full column list. The /api/v2/meta/tables/{id}
# endpoint returns 404 on NocoDB v2026.04.x; /api/v1/db/meta/tables/{id}
# is the working equivalent and returns the same shape (id, title, columns).
table = _api(sess, base_url, token, "GET", f"/api/v1/db/meta/tables/{table['id']}")
col = find_column(table, "Message_ID")
if not col:
if args.check:
print(f" MISSING column Message_ID in {module['title']!r}", file=sys.stderr)
any_drift = True
continue
elif not col.get("un"):
if args.check:
print(f" MISSING unique constraint on Message_ID in {module['title']!r}",
file=sys.stderr)
any_drift = True
else:
if ensure_unique_on_message_id(sess, base_url, token, col):
print(f" added unique constraint on {module['title']!r}.Message_ID")
if not args.check:
# Re-fetch and add any columns that are still missing
table = _api(sess, base_url, token, "GET", f"/api/v1/db/meta/tables/{table['id']}")
existing_names = {c.get("title") or c.get("column_name")
for c in table.get("columns", [])}
missing = [c for c in module["columns"] if c["name"] not in existing_names]
missing += [c for c in COMMON_COLUMNS if c["name"] not in existing_names]
if missing:
added = add_missing_columns(sess, base_url, token, table, missing)
print(f" added {added} missing column(s) to {module['title']!r}")
module_table_ids.append((module["name"], table["id"]))
# Audit-trail tables (REQ-INB-19+). Same idempotent create/check
# path as the module tables, but the columns are table-specific
# (no COMMON_COLUMNS — papra_index has its own DateTime, kind, etc.).
audit_table_ids: list[tuple[str, str]] = []
for audit in AUDIT_TABLES:
table = find_table(sess, base_url, token, base_id, audit["title"])
if not table:
if args.check:
print(f" MISSING audit table {audit['title']!r}", file=sys.stderr)
any_drift = True
continue
table = create_table(sess, base_url, token, base_id, audit)
print(f" created audit table {audit['title']!r} id={table['id']}")
else:
print(f" found audit table {audit['title']!r} id={table['id']}")
table = _api(sess, base_url, token, "GET", f"/api/v1/db/meta/tables/{table['id']}")
col = find_column(table, "Message_ID")
if not col:
if args.check:
print(f" MISSING column Message_ID in audit {audit['title']!r}",
file=sys.stderr)
any_drift = True
continue
elif not col.get("un"):
if args.check:
print(f" MISSING unique constraint on Message_ID in audit "
f"{audit['title']!r}", file=sys.stderr)
any_drift = True
else:
if ensure_unique_on_message_id(sess, base_url, token, col):
print(f" added unique constraint on audit "
f"{audit['title']!r}.Message_ID")
if not args.check:
table = _api(sess, base_url, token, "GET",
f"/api/v1/db/meta/tables/{table['id']}")
existing_names = {c.get("title") or c.get("column_name")
for c in table.get("columns", [])}
missing = [c for c in audit["columns"] if c["name"] not in existing_names]
if missing:
added = add_missing_columns(sess, base_url, token, table, missing)
print(f" added {added} missing column(s) to audit "
f"{audit['title']!r}")
audit_table_ids.append((audit["name"], table["id"]))
print()
print("NOCODB_MODULE_TABLES (set this as a Gitea repo secret):")
print(json.dumps({name: tid for name, tid in module_table_ids}, indent=2))
print()
print("Pipeline env var block to paste into Gitea repo secrets:")
print(" NOCODB_BASE_URL=" + base_url)
print(" NOCODB_API_TOKEN=<redacted>")
print(f" NOCODB_PROJECT_ID={base_id}")
print(f" NOCODB_MODULE_TABLES={json.dumps({n: t for n, t in module_table_ids})}")
# v1.4 REQ-INB-41: seed curated NocoDB views per module
if args.seed_views and not args.check:
seed_curated_views(sess, base_url, token, module_table_ids)
if args.check and any_drift:
print("\n--check found drift; rerun without --check to fix.", file=sys.stderr)
return 1
return 0
# ── v1.4 REQ-INB-41: curated views ──────────────────────
CURATED_VIEWS = [
{
"table": "finance",
"title": "Open invoices 30d",
"filter": "(status,eq,open)~and(due_date,le,today+30d)",
},
{
"table": "crm",
"title": "New contacts 7d",
"filter": "(created_at,ge,today-7d)",
},
{
"table": "intel",
"title": "Intel watchlist alerts",
"filter": "(priority,eq,high)",
},
{
"table": "promotions",
"title": "Promotions opt-out queue",
"filter": "(opt_out,eq,true)",
},
]
def seed_curated_views(sess, base_url, token, module_table_ids):
"""Seed 4 curated NocoDB views, one per module. Idempotent."""
from urllib.parse import quote
table_id_by_name = {n: tid for n, tid in module_table_ids}
for view in CURATED_VIEWS:
table_id = table_id_by_name.get(view["table"])
if not table_id:
print(f" skip view {view['title']!r} — table {view['table']!r} not found")
continue
try:
view_body = {
"title": view["title"],
"type": "grid",
"filters": view["filter"],
}
_api(sess, base_url, token, "POST",
f"/api/v2/tables/{table_id}/views", json_body=view_body)
print(f" seeded view {view['title']!r} on {view['table']!r}")
except Exception as exc:
print(f" WARN: view {view['title']!r} seed failed: {exc}", file=sys.stderr)
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink