Exploring schema evolution with ontology-driven propagation

I'm building a DuckDB analytics view over a fintech dataset. The schema evolves at any time — new columns can arrive without warning. I need an ontology that encodes the access policy as plain-English subject — predicate — object triples. No CamelCase, no identifier syntax — write it the way you'd write rules for a colleague who hasn't seen the codebase. Group claims under a shared subject with bullets when the subject repeats. Collapse parallel claims into a single list when the same predicate applies to several subjects. Keep the file short and scannable.

The ontology will be applied two ways: a deterministic interpreter handles crisp cases (name patterns, dtypes, cardinality), and an LLM handles ambiguous cases by sampling values. Write rules that work for both readers.

Keep it concise, essential rules only, no exhaustive type enumerations.

Define:
- The view's goal: expose analytics-safe columns, exclude PII and high-cardinality text unless the content is verifiably non-personal, adapt automatically on schema change
- Which column classes pass and which are rejected
- PII fragments — representative examples across name, contact, financial, and identity categories
- A cardinality threshold of 3% (a text column is high-cardinality if more than 3 in 100 values are unique)
- A runtime escalation rule: for high-cardinality text whose name doesn't match a PII fragment, inspect a sample of values; pass if values are identifiers, codes, or non-personal descriptions; reject if values contain personal information
- That new columns are evaluated automatically by the same rules with no code change needed

The view:
- exposes analytics-safe columns
- excludes PII columns
- excludes high-cardinality text columns unless their values are verifiably non-personal
- adapts automatically when the schema changes

Columns that pass: numeric, boolean, date, low-cardinality text.
Columns that are rejected: PII, high-cardinality text containing personal information.

A column is PII if its name contains a fragment like:
- name, full_name, first_name, last_name, username
- email, email_address, phone, phone_number, mobile, address, city, zip, postal_code
- card_number, iban, account_number, routing_number, cvv, pan
- ssn, passport, national_id, tax_id, dob, date_of_birth, ip_address, device_id

A text column is high-cardinality if more than 3% of its values are unique.

For high-cardinality text columns whose name does not match a PII fragment:
- inspect a sample of values
- the column passes if its values are identifiers, codes, machine-generated strings, or non-personal descriptions
- the column is rejected if its values contain personal information: names, contact details, financial identifiers, free-form text describing individuals, or anything resembling the PII categories above

New columns:
- are evaluated by the same rules
- require no code change to be classified
- inherit the policy of their column class

import json
import anthropic

client = anthropic.Anthropic(api_key=api_key.value)
MODEL = "claude-sonnet-4-6"

def llm_decide(name, dtype, cardinality, sample):
    # Retry loop for model stability
    for _ in range(3):
        try:
            response = client.messages.create(
                model=MODEL,
                max_tokens=200,
                system=(
                    f"You enforce this data access policy:\n\n{ontology_text}\n\n"
                    "Decide if a column is analytics-safe. Reply with JSON only, no markdown: "
                    '{"decision": "pass" or "reject", "reason": "<one sentence>"}'
                ),
                messages=[{
                    "role": "user",
                    "content": (
                        f"Column name: {name}\n"
                        f"Dtype: {dtype}\n"
                        f"Cardinality ratio: {cardinality:.1%}\n"
                        f"Sample values: {sample}"
                    ),
                }],
            )
            out = json.loads(response.content[0].text)
            
            # Ensure the model sticks to the allowed ontology decisions
            if out.get("decision") in ["pass", "reject"]:
                return out["decision"], out["reason"]
        except:
            print(f"  Retrying {name} due to invalid LLM output...")
            continue

    # Fail-closed posture for safety
    return "reject", "classification_failed_after_retries"

def build_safe_view(pipeline_name: str, raw_table: str, safe_view: str):
    con = dlt.attach(pipeline_name).dataset().ibis()
    tbl = con.table(raw_table)
    total = tbl.count().execute()

    safe_columns = []
    for col_name, dtype in zip(tbl.schema().names, tbl.schema().types):
        if col_name.startswith("_dlt_") or col_name == "_id":
            continue
        cardinality = tbl[col_name].nunique().execute() / total
        sample = tbl.select(col_name).limit(20).execute()[col_name].tolist()
        decision, reason = llm_decide(col_name, dtype, cardinality, sample)
        marker = "✓ PASS  " if decision == "pass" else "✗ REJECT"
        print(f"  {marker}  {col_name}  ← {reason}")
        if decision == "pass":
            safe_columns.append(col_name)

    con.create_view(safe_view, tbl.select(safe_columns), overwrite=True)
    print(f"\nVIEW '{safe_view}' created — {len(safe_columns)} columns: {', '.join(safe_columns)}")