"""The relation workspace the relation, search, vocabulary and correction examples run on. The concept pages under ``docs/integrator/`` run against a seeded annual report (numeric facts). Relations need a second workspace: one with a PACK applied (the relations ``held_role`` / ``screening`` / ``engagement`` and their vocabulary) and an EXPORT connected as relation rows. This module builds it, and every step but one goes through the same public verbs an agent or an operator calls: 1. create a workspace owned by the key's organisation — the one step made directly instead of over HTTP: the examples run in development, where a workspace created over HTTP belongs to no organisation, with a key or without one (in every other environment ``POST /api/workspaces`` asks for a key and binds the workspace to that key's organisation — see the quickstart); 2. ``POST /v1/public/apply_pack`` with ``bundle_expert_network``; 3. write the two tables below into an SQLite file — the "warehouse export"; 4. ``POST /v1/public/connect`` twice, with a ``relation_mapping`` per table. ``connect`` refuses every ``sqlite:///`` URL unless an OPERATOR has listed its directory in ``EPISTEM_CONNECT_ALLOWED_SQLITE_DIRS``. The examples run inside the test process, so :func:`connect` sets that variable for the one call and puts it back — the deployment equivalent is one line of server environment (``docs/operator/connect.md``). The steps deliberately mirror ``demo/expert_network/load_structured.py`` (``ensure_pack``, ``_post_connect``) and the connect helpers of the tests, over a small invented dataset a reader can hold in their head; this module exists so the documentation imports nothing from a demo or a test. Every name below is invented for the documentation. ``roster(client)`` is SHARED and read-only by convention (built once per app); an example that corrects, re-syncs or erases takes ``new_roster(client)`` so no other example sees what it changed. """ from __future__ import annotations import atexit import os import shutil import sqlite3 import tempfile import weakref from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Tuple #: The signed pack the workspace's relations come from. PACK_BUNDLE_ID = "bundle_expert_network" #: The environment variable the connect verb's datasource policy reads. ALLOWED_SQLITE_DIRS_ENV = "EPISTEM_CONNECT_ALLOWED_SQLITE_DIRS" #: The month the export was pulled: the witness an OPEN role (``period_to`` NULL) needs. EXPORT_AS_OF = "2024-06" ROLE_COLUMNS: Tuple[str, ...] = ("row_id", "person", "company", "title", "seniority", "function", "geography", "period_from", "period_to", "screened") #: One row per held role. ``screened`` says whether this export's system looked at the #: person's screenings: ``yes`` writes a screening COVERAGE row, NULL writes none. ROLES: List[Tuple[Any, ...]] = [ (1, "halvard_quist", "Kestrel Pumps AG", "Vice President, Sales", "VP", "Sales", "Germany", "2016-03", "2019-01", "yes"), (2, "halvard_quist", "Lumen Ridge", "Director, Sales", "Director", "Sales", "Austria", "2019-01", None, "yes"), (3, "isolde_marbrook", "Corvid Analytics", "Senior Vice President, Sales", "SVP", "Sales", "Switzerland", "2018-06", "2022-02", "yes"), (4, "odile_fenwright", "Harrow & Finch", "Director, Marketing", "Director", "Marketing", "Germany", "2014-01", "2019-02", "yes"), (5, "teodor_brask", "Sable Point", "Chief Operating Officer", "C-level", "Operations", "United States", "2017-01", "2023-01", "yes"), (6, "yusuf_alderane", "Pellucid Freight", "Vice President, Sales", "VP", "Sales", "Austria", "2020-04", None, None), ] SCREENING_COLUMNS: Tuple[str, ...] = ("screening_id", "person", "kind", "outcome", "screened_on") SCREENINGS: List[Tuple[Any, ...]] = [ (1, "halvard_quist", "compliance", "passed", "2024-02"), (2, "isolde_marbrook", "compliance", "failed", "2024-03"), (3, "odile_fenwright", "compliance", "passed", "2023-11"), ] ROLES_MAPPING: Dict[str, Any] = { "relation": "held_role", "entity_column": "person", "slot_columns": {"company": "company", "title": "title", "seniority": "seniority", "function": "function", "geography": "geography"}, "key_columns": ["row_id"], "period_from_column": "period_from", "period_to_column": "period_to", "as_of": EXPORT_AS_OF, "coverage_columns": {"screening": "screened"}, } SCREENINGS_MAPPING: Dict[str, Any] = { "relation": "screening", "entity_column": "person", "slot_columns": {"kind": "kind", "outcome": "outcome", "as_of": "screened_on"}, "key_columns": ["screening_id"], } @dataclass(frozen=True) class Roster: """What :func:`new_roster` built: the workspace, the two connected sources, and the SQLite export file (an example that re-syncs edits it and connects again).""" workspace_id: str roles_source: str screenings_source: str export: Path @property def export_url(self) -> str: return f"sqlite:///{self.export}" _EXPORT_DIRS: List[Path] = [] def _remove_export_dirs() -> None: for directory in _EXPORT_DIRS: shutil.rmtree(directory, ignore_errors=True) atexit.register(_remove_export_dirs) def export_dir() -> Path: """A fresh directory for a warehouse export, removed when the process exits — not before: a connected table's receipts are re-read from the file for as long as the process serves them.""" directory = Path(tempfile.mkdtemp(prefix="limma_docs_export_")) _EXPORT_DIRS.append(directory) return directory def write_export(directory: Path) -> Path: """The two tables, written to ``/roster_export.db``.""" path = Path(directory) / "roster_export.db" with sqlite3.connect(path) as conn: conn.execute("CREATE TABLE roles (row_id INTEGER PRIMARY KEY, person TEXT, " "company TEXT, title TEXT, seniority TEXT, function TEXT, " "geography TEXT, period_from TEXT, period_to TEXT, screened TEXT)") conn.executemany(f"INSERT INTO roles VALUES ({', '.join('?' * len(ROLE_COLUMNS))})", ROLES) conn.execute("CREATE TABLE screenings (screening_id INTEGER PRIMARY KEY, " "person TEXT, kind TEXT, outcome TEXT, screened_on TEXT)") conn.executemany( f"INSERT INTO screenings VALUES ({', '.join('?' * len(SCREENING_COLUMNS))})", SCREENINGS) return path def connect(client: Any, workspace_id: str, export: Path, table: str, relation_mapping: Dict[str, Any], replaces: str = "") -> Any: """``POST /v1/public/connect`` for one table of ``export``, with the operator's sqlite allowlist naming the export's directory for the duration of the call.""" body: Dict[str, Any] = {"workspace_id": workspace_id, "database_url": f"sqlite:///{export}", "table": table, "relation_mapping": relation_mapping} if replaces: body["replaces"] = replaces previous = os.environ.get(ALLOWED_SQLITE_DIRS_ENV) os.environ[ALLOWED_SQLITE_DIRS_ENV] = str(Path(export).parent) try: return client.post("/v1/public/connect", json=body) finally: if previous is None: os.environ.pop(ALLOWED_SQLITE_DIRS_ENV, None) else: os.environ[ALLOWED_SQLITE_DIRS_ENV] = previous def _checked(response: Any, what: str) -> Dict[str, Any]: if response.status_code != 200: raise RuntimeError(f"{what} answered {response.status_code}: {response.text}") return response.json() def _organisation(client: Any) -> str: """The organisation of the key ``client`` sends, resolved by the API's own resolver (``resolve_presented_key``) against the configured ``EPISTEM_API_KEYS``.""" from epistem.api.auth.api_key import resolve_presented_key record = resolve_presented_key(client.headers.get("X-API-Key")) if record is None: raise RuntimeError("the client's X-API-Key is not a configured key") return record.org_id def new_workspace(client: Any, *, name: str = "docs-example") -> str: """An EMPTY workspace owned by the organisation of ``client``'s key — step 1 above, and nothing else: no pack, no source.""" store = client.app.state.workspace_manager.store return store.create_workspace(name=name, org_id=_organisation(client)).workspace_id def new_roster(client: Any, *, name: str = "roster") -> Roster: """A fresh workspace with the pack applied and both tables connected.""" workspace_id = new_workspace(client, name=name) _checked(client.post("/v1/public/apply_pack", json={ "workspace_id": workspace_id, "bundle_id": PACK_BUNDLE_ID}), "apply_pack") export = write_export(export_dir()) roles = _checked(connect(client, workspace_id, export, "roles", ROLES_MAPPING), "connect roles") screenings = _checked(connect(client, workspace_id, export, "screenings", SCREENINGS_MAPPING), "connect screenings") return Roster(workspace_id=workspace_id, roles_source=roles["source_id"], screenings_source=screenings["source_id"], export=export) _SHARED: "weakref.WeakKeyDictionary[Any, Roster]" = weakref.WeakKeyDictionary() def roster(client: Any) -> Roster: """The shared roster of ``client``'s app — built on first use, never mutated.""" app = client.app if app not in _SHARED: _SHARED[app] = new_roster(client, name="roster-shared") return _SHARED[app] def update_export(export: Path, sql: str, params: Sequence[Any] = ()) -> None: """Change the warehouse export in place — what a nightly re-sync then reads.""" with sqlite3.connect(export) as conn: conn.execute(sql, tuple(params)) __all__ = ["PACK_BUNDLE_ID", "ROLES", "SCREENINGS", "ROLES_MAPPING", "SCREENINGS_MAPPING", "Roster", "roster", "new_roster", "new_workspace", "connect", "export_dir", "write_export", "update_export"]