209 lines
9.6 KiB
Python
209 lines
9.6 KiB
Python
"""JSON/SQLite registry of semantic relation schemas and scoped slot occurrences.
|
|
|
|
A role definition may be reused deliberately. Occurrences remain local to a
|
|
simplex: sharing a role definition never equates the objects bound to its aliases.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import json
|
|
import sqlite3
|
|
from collections.abc import Iterable, Mapping
|
|
from pathlib import Path
|
|
from urllib.parse import quote
|
|
|
|
from .schema import KIND_GOAL, KIND_OWN, KIND_SIMPLEX, Role, Simplex, role_from_dict
|
|
|
|
|
|
def slot_key(parent: str, alias: str) -> str:
|
|
"""An occurrence identity, distinct even when two slots reuse one definition."""
|
|
return f"{parent}/slot/{quote(alias, safe='')}"
|
|
|
|
|
|
class SchemaRegistry:
|
|
"""Immutable versioned records; reusing a key with changed content is rejected."""
|
|
|
|
def __init__(self, path: str | Path = ":memory:") -> None:
|
|
self.connection = sqlite3.connect(str(path))
|
|
self.connection.execute("PRAGMA foreign_keys = ON")
|
|
self.connection.executescript("""
|
|
CREATE TABLE IF NOT EXISTS schema_nodes (
|
|
key TEXT PRIMARY KEY,
|
|
kind TEXT NOT NULL CHECK(kind IN ('role', 'simplex')),
|
|
payload TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS schema_slots (
|
|
parent TEXT NOT NULL REFERENCES schema_nodes(key) ON DELETE CASCADE,
|
|
alias TEXT NOT NULL,
|
|
child TEXT NOT NULL REFERENCES schema_nodes(key) ON DELETE RESTRICT,
|
|
position INTEGER NOT NULL,
|
|
PRIMARY KEY(parent, alias),
|
|
UNIQUE(parent, position)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS schema_slots_child ON schema_slots(child);
|
|
""")
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
self.close()
|
|
|
|
def close(self) -> None:
|
|
self.connection.close()
|
|
|
|
def _insert(self, key: str, kind: str, payload: dict) -> None:
|
|
if not isinstance(key, str) or not key:
|
|
raise ValueError("a registry key must be nonempty")
|
|
if (kind == "simplex") != (payload.get("kind") == KIND_SIMPLEX):
|
|
raise ValueError("nested simplex roles must use a Simplex object")
|
|
encoded = json.dumps(payload, sort_keys=True, ensure_ascii=False, allow_nan=False)
|
|
existing = self.connection.execute(
|
|
"SELECT kind, payload FROM schema_nodes WHERE key = ?", (key,)).fetchone()
|
|
if existing is not None and existing != (kind, encoded):
|
|
raise ValueError(f"key {key!r} already has different content; use a new version key")
|
|
self.connection.execute("INSERT OR IGNORE INTO schema_nodes VALUES (?, ?, ?)",
|
|
(key, kind, encoded))
|
|
|
|
def set_role(self, role: Role, key: str | None = None) -> str:
|
|
if isinstance(role, Simplex):
|
|
raise ValueError("use set_simplex for nested schemas")
|
|
key = key or f"role:{role.name}"
|
|
with self.connection:
|
|
self._insert(key, "role", role.as_dict())
|
|
return key
|
|
|
|
def set_simplex(self, simplex: Simplex, key: str | None = None,
|
|
role_keys: Mapping[str, str] | None = None) -> str:
|
|
"""Default children are scoped. role_keys explicitly opts into definition reuse."""
|
|
key = key or f"schema:{simplex.name}"
|
|
role_keys = dict(role_keys or {})
|
|
if not set(role_keys) <= set(simplex.roles):
|
|
raise ValueError("reuse map has an unknown alias")
|
|
with self.connection:
|
|
self._put_simplex(simplex, key, role_keys, set())
|
|
return key
|
|
|
|
def _put_simplex(self, simplex: Simplex, key: str, reuse: dict[str, str],
|
|
active: set[int]) -> None:
|
|
if id(simplex) in active:
|
|
raise ValueError("cyclic simplex nesting")
|
|
active = active | {id(simplex)}
|
|
# Metadata only; children live in a relational slot table, not pickle blobs.
|
|
payload = Role.as_dict(simplex)
|
|
existing = self.connection.execute(
|
|
"SELECT key FROM schema_nodes WHERE key = ?", (key,)).fetchone()
|
|
if existing:
|
|
restored = self.get_simplex(key)
|
|
if restored is None or restored.as_dict() != simplex.as_dict():
|
|
raise ValueError(f"key {key!r} already has different content; use a new version key")
|
|
expected = {alias: reuse.get(alias, slot_key(key, alias)) for alias in simplex.roles}
|
|
if self.slots(key) != expected:
|
|
raise ValueError("same schema key cannot change its role-definition references")
|
|
return
|
|
self._insert(key, "simplex", payload)
|
|
for position, (alias, child) in enumerate(simplex.roles.items()):
|
|
child_key = reuse.get(alias, slot_key(key, alias))
|
|
if alias in reuse:
|
|
restored = self.get(child_key)
|
|
if restored is None or restored.as_dict() != child.as_dict():
|
|
raise ValueError(f"reused role definition does not match slot {alias!r}")
|
|
elif isinstance(child, Simplex):
|
|
self._put_simplex(child, child_key, {}, active)
|
|
else:
|
|
self._insert(child_key, "role", child.as_dict())
|
|
self.connection.execute("INSERT INTO schema_slots VALUES (?, ?, ?, ?)",
|
|
(key, alias, child_key, position))
|
|
|
|
def get(self, key: str) -> Role | None:
|
|
return self._get(key, set())
|
|
|
|
def _get(self, key: str, active: set[str]) -> Role | None:
|
|
if key in active:
|
|
raise ValueError("cyclic schema references in registry")
|
|
row = self.connection.execute(
|
|
"SELECT kind, payload FROM schema_nodes WHERE key = ?", (key,)).fetchone()
|
|
if row is None:
|
|
return None
|
|
kind, encoded = row
|
|
payload = json.loads(encoded)
|
|
if kind == "role":
|
|
return role_from_dict(payload)
|
|
slots = self.slots(key)
|
|
children = {alias: self._get(child, active | {key}) for alias, child in slots.items()}
|
|
return Simplex(payload["name"], children, context=payload.get("context", ""),
|
|
attrs=payload.get("attrs"))
|
|
|
|
def get_role(self, key: str) -> Role | None:
|
|
value = self.get(key)
|
|
return value if not isinstance(value, Simplex) else None
|
|
|
|
def get_simplex(self, key: str) -> Simplex | None:
|
|
value = self.get(key)
|
|
return value if isinstance(value, Simplex) else None
|
|
|
|
def slots(self, key: str) -> dict[str, str]:
|
|
return dict(self.connection.execute(
|
|
"SELECT alias, child FROM schema_slots WHERE parent = ? ORDER BY position", (key,)))
|
|
|
|
def keys(self, pattern: str = "*") -> list[str]:
|
|
return [key for (key,) in self.connection.execute("SELECT key FROM schema_nodes ORDER BY key")
|
|
if fnmatch.fnmatchcase(key, pattern)]
|
|
|
|
def search_roles(self, query: str, limit: int = 10) -> list[dict]:
|
|
"""Literal substring candidates; search does not authorize semantic reuse."""
|
|
if not query.strip() or limit < 1:
|
|
return []
|
|
found = []
|
|
for key, payload in self.connection.execute(
|
|
"SELECT key, payload FROM schema_nodes WHERE kind = 'role' ORDER BY key"):
|
|
if query.casefold() in (key + payload).casefold():
|
|
found.append({"key": key, "role": json.loads(payload)})
|
|
if len(found) == limit:
|
|
break
|
|
return found
|
|
|
|
def find_by_roles(self, role_keys: Iterable[str], match_all: bool = False) -> list[str]:
|
|
"""Find schema keys containing definition keys, including nested descendants."""
|
|
wanted = set(role_keys)
|
|
if not wanted:
|
|
return []
|
|
found = []
|
|
for key in self._simplex_keys():
|
|
descendants = {row[0] for row in self.connection.execute("""
|
|
WITH RECURSIVE descendants(child) AS (
|
|
SELECT child FROM schema_slots WHERE parent = ?
|
|
UNION
|
|
SELECT slots.child FROM schema_slots slots JOIN descendants d ON slots.parent = d.child
|
|
) SELECT child FROM descendants
|
|
""", (key,))}
|
|
matches = wanted <= descendants if match_all else bool(wanted & descendants)
|
|
if matches:
|
|
found.append(key)
|
|
return found
|
|
|
|
def find_solvable(self, known_slots: Iterable[str]) -> list[dict]:
|
|
"""Structural readiness by scoped occurrences, not proof or value unification."""
|
|
known = set(known_slots)
|
|
candidates = []
|
|
for key in self._simplex_keys():
|
|
schema = self.get_simplex(key)
|
|
inputs, outputs = [], []
|
|
for alias, role in schema.roles.items():
|
|
target = outputs if role.kind in (KIND_GOAL, KIND_OWN) else inputs
|
|
target.append(slot_key(key, alias))
|
|
missing_outputs = [output for output in outputs if output not in known]
|
|
if missing_outputs and set(inputs) <= known:
|
|
candidates.append({"schema_key": key, "input_slots": inputs,
|
|
"target_slots": missing_outputs, "structural_candidate_only": True})
|
|
return candidates
|
|
|
|
def _simplex_keys(self) -> list[str]:
|
|
return [row[0] for row in self.connection.execute(
|
|
"SELECT key FROM schema_nodes WHERE kind = 'simplex' ORDER BY key")]
|
|
|
|
def info(self) -> dict[str, int]:
|
|
counts = dict(self.connection.execute("SELECT kind, count(*) FROM schema_nodes GROUP BY kind"))
|
|
return {"roles": counts.get("role", 0), "simplices": counts.get("simplex", 0),
|
|
"slots": self.connection.execute("SELECT count(*) FROM schema_slots").fetchone()[0]}
|