semantica/cookbook/advanced/13_Manual_Ontology_Snowflake_Mapping.ipynb

Manual Ontology + Snowflake Mapping

This notebook answers a specific workflow:

"I want to design the ontology myself — not have AI infer it from my tables — and then map Snowflake data to it explicitly."

What this notebook demonstrates

Step	What happens	Who controls it
1	Design ontology classes and properties	You (Python dict)
2	Model n-ary facts with reification	You (AssociativeClassBuilder)
3	Pull rows from Snowflake	Semantica SnowflakeIngestor
4	Map columns → ontology-aligned graph	You (explicit transform)
5	Validate + export OWL / SHACL	Semantica OntologyEngine
6	Load to triplet store and query	Semantica TripletStore

What this notebook does NOT do

• No LLM-driven ontology generation
• No schema introspection or table-to-class inference
• No "suggest ontology from my data"

Standards coverage

Feature	Status
OWL 2 (Turtle / RDF-XML)	Supported
SHACL 1.1 shapes	Supported
SPARQL 1.1	Supported
Reification / n-ary facts	Supported via AssociativeClassBuilder
SPARQL 1.2 (reifier annotation, LATERAL)	Planned
SHACL 1.2 (sh:severity extensions, SHACL-AF)	Planned

In [ ]:

!pip install -qU semantica

In [ ]:

import os
from typing import Any, Dict, List

from semantica.ingest import SnowflakeIngestor
from semantica.kg.methods import build_kg
from semantica.ontology import AssociativeClassBuilder, OntologyEngine
from semantica.triplet_store import TripletStore

Step 1: Hand-Design the Ontology in Python

You define every class and property explicitly. Nothing is read from Snowflake at this stage.

Design decisions that belong to you:

• Which classes exist and what they mean
• Which properties are datatype vs. object properties
• Domain, range, and cardinality constraints
• Which properties are required (later enforced by SHACL)

This dict versions with your code. It does not change when your database schema changes.

In [ ]:

BASE_URI = "https://example.com/hr/"

# Your ontology — designed by you, not inferred by Semantica.
ontology: Dict[str, Any] = {
    "name": "EmploymentDomainOntology",
    "uri": f"{BASE_URI}EmploymentDomainOntology",
    "namespace": {"base_uri": BASE_URI},

    # You decide the class taxonomy
    "classes": [
        {"name": "Person",          "uri": f"{BASE_URI}Person"},
        {"name": "Organization",    "uri": f"{BASE_URI}Organization"},
        {"name": "Role",            "uri": f"{BASE_URI}Role"},
        # EmploymentEvent is a reification node.
        # It connects Person + Organization + Role and carries salary/date context.
        {"name": "EmploymentEvent", "uri": f"{BASE_URI}EmploymentEvent"},
    ],

    # Each property carries a full URI so TripletStore stores it as hr:<name>
    # rather than the default urn:property:<name>.
    # This ensures SPARQL queries using PREFIX hr: match what is actually stored.
    "properties": [
        # Datatype properties
        {"name": "name",      "uri": f"{BASE_URI}name",      "type": "datatype", "domain": "Person",          "range": "string",  "required": True},
        {"name": "legalName", "uri": f"{BASE_URI}legalName", "type": "datatype", "domain": "Organization",    "range": "string",  "required": True},
        {"name": "title",     "uri": f"{BASE_URI}title",     "type": "datatype", "domain": "Role",            "range": "string",  "required": True},
        {"name": "startDate", "uri": f"{BASE_URI}startDate", "type": "datatype", "domain": "EmploymentEvent", "range": "date"},
        {"name": "endDate",   "uri": f"{BASE_URI}endDate",   "type": "datatype", "domain": "EmploymentEvent", "range": "date"},
        {"name": "salary",    "uri": f"{BASE_URI}salary",    "type": "datatype", "domain": "EmploymentEvent", "range": "decimal"},

        # Object properties — reification spokes (required)
        {"name": "employee",  "uri": f"{BASE_URI}employee",  "type": "object",   "domain": "EmploymentEvent", "range": "Person",       "required": True},
        {"name": "employer",  "uri": f"{BASE_URI}employer",  "type": "object",   "domain": "EmploymentEvent", "range": "Organization", "required": True},
        {"name": "role",      "uri": f"{BASE_URI}role",      "type": "object",   "domain": "EmploymentEvent", "range": "Role",         "required": True},

        # Shortcut edges — direct person→org / person→role without traversing the event node
        {"name": "worksFor",  "uri": f"{BASE_URI}worksFor",  "type": "object",   "domain": "Person",          "range": "Organization"},
        {"name": "hasRole",   "uri": f"{BASE_URI}hasRole",   "type": "object",   "domain": "Person",          "range": "Role"},
    ],
}

ontology

Step 2: Reification — Modeling N-Ary Facts

The problem with binary triples: A simple triple (Alice, worksFor, Acme) cannot carry extra context such as salary, start date, or role. Standard RDF reification and OWL n-ary patterns solve this by introducing an intermediate node.

Semantica's AssociativeClassBuilder is the Pythonic API for this pattern:

EmploymentEvent
    ├── employee  → Person          (required)
    ├── employer  → Organization    (required)
    ├── role      → Role            (required)
    ├── startDate → xsd:date
    ├── endDate   → xsd:date
    └── salary    → xsd:decimal

On SPARQL 1.1 vs. SPARQL 1.2:

• SPARQL 1.1 (current): traverse the event node explicitly — ?event hr:employee ?person ; hr:salary ?salary
• SPARQL 1.2 (planned): the draft reifier annotation syntax allows attaching context to triples directly, without a separate intermediate node. Semantica will adopt this once the spec is ratified.

On SHACL 1.1 vs. SHACL 1.2:

• SHACL 1.1 (current): sh:NodeShape + sh:PropertyShape constraints are exported for all required properties and enforced at load time.
• SHACL 1.2 (planned): sh:severity profile extensions and SHACL-AF rules are on the roadmap.

In [ ]:

assoc_builder = AssociativeClassBuilder()

employment_assoc = assoc_builder.create_associative_class(
    name="EmploymentEvent",
    connects=["Person", "Organization", "Role"],
    temporal=True,  # adds startDate / endDate handling
    properties={
        "startDate": "xsd:date",
        "endDate":   "xsd:date",
        "salary":    "xsd:decimal",
    },
)

validation_result = assoc_builder.validate_associative_class(employment_assoc)

# AssociativeClass is a dataclass — use attribute access, not .get()
print("AssociativeClass structure:")
print(f"  name:       {employment_assoc.name}")
print(f"  connects:   {employment_assoc.connects}")
print(f"  temporal:   {employment_assoc.temporal}")
print(f"  properties: {list(employment_assoc.properties.keys())}")
print(f"\nValidation passed: {validation_result}")

Step 3: Ingest Snowflake Rows (Extraction Only)

SnowflakeIngestor retrieves rows — nothing more. It does not:

• Inspect your table schema
• Suggest classes or properties
• Infer relationships from column names

Set USE_LIVE_SNOWFLAKE=true plus the env vars below to connect to a real warehouse. Otherwise the stub data is used.

In [ ]:

def fetch_rows_from_snowflake() -> List[Dict[str, Any]]:
    if os.getenv("USE_LIVE_SNOWFLAKE", "false").lower() != "true":
        return [
            {
                "EMPLOYEE_ID":   "E100",
                "EMPLOYEE_NAME": "Alice Johnson",
                "ORG_ID":        "O10",
                "ORG_NAME":      "Acme Corp",
                "ROLE_ID":       "R7",
                "ROLE_TITLE":    "Senior Engineer",
                "START_DATE":    "2025-01-15",
                "END_DATE":      None,
                "SALARY":        160000,
            },
            {
                "EMPLOYEE_ID":   "E101",
                "EMPLOYEE_NAME": "Bob Singh",
                "ORG_ID":        "O10",
                "ORG_NAME":      "Acme Corp",
                "ROLE_ID":       "R9",
                "ROLE_TITLE":    "Data Architect",
                "START_DATE":    "2024-09-01",
                "END_DATE":      None,
                "SALARY":        185000,
            },
        ]

    ingestor = SnowflakeIngestor(
        account=os.getenv("SNOWFLAKE_ACCOUNT"),
        user=os.getenv("SNOWFLAKE_USER"),
        password=os.getenv("SNOWFLAKE_PASSWORD"),
        warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
        database=os.getenv("SNOWFLAKE_DATABASE"),
        schema=os.getenv("SNOWFLAKE_SCHEMA", "PUBLIC"),
    )
    query = (
        "SELECT EMPLOYEE_ID, EMPLOYEE_NAME, "
        "ORG_ID, ORG_NAME, ROLE_ID, ROLE_TITLE, "
        "START_DATE, END_DATE, SALARY "
        "FROM HR_EMPLOYMENT_FACT"
    )
    data = ingestor.ingest_query(query)
    ingestor.close()
    return data.data


rows = fetch_rows_from_snowflake()
rows[:2]

Step 4: Map Rows to Ontology Concepts Explicitly

This is the semantic transformation layer — the part that makes your ontology real.

Semantica does not guess which column becomes which entity or property. Every assignment is code you write and own:

• Stable node IDs — deterministic, collision-safe, derived from business keys
• Class assignment — matches what you declared in Step 1
• Property routing — each column value goes to the correct ontology property
• Reification wiring — EmploymentEvent is linked to its three participants

When your Snowflake schema changes, only this function needs updating. The ontology stays stable.

In [ ]:

def map_rows_to_kg(rows: List[Dict[str, Any]]) -> Dict[str, Any]:
    entities: Dict[str, Dict[str, Any]] = {}
    relationships: List[Dict[str, Any]] = []

    for row in rows:
        # Stable, deterministic node IDs derived from business keys
        person_id = f"person:{row['EMPLOYEE_ID']}"
        org_id    = f"org:{row['ORG_ID']}"
        role_id   = f"role:{row['ROLE_ID']}"
        # Event ID includes all three participants + start date so that
        # a re-hired employee gets a distinct event node, not an overwrite.
        event_id  = f"employment:{row['EMPLOYEE_ID']}:{row['ORG_ID']}:{row['START_DATE']}"

        # Entities — "type" must match a class name from Step 1
        entities[person_id] = {
            "id": person_id,
            "type": "Person",
            "properties": {"name": row["EMPLOYEE_NAME"]},
        }
        entities[org_id] = {
            "id": org_id,
            "type": "Organization",
            "properties": {"legalName": row["ORG_NAME"]},
        }
        entities[role_id] = {
            "id": role_id,
            "type": "Role",
            "properties": {"title": row["ROLE_TITLE"]},
        }

        # Reification node — filter out None values so TripletStore does not
        # stringify None as the literal "None" for open-ended employment.
        event_props = {
            "startDate": row["START_DATE"],
            "endDate":   row["END_DATE"],
            "salary":    row["SALARY"],
        }
        entities[event_id] = {
            "id": event_id,
            "type": "EmploymentEvent",
            "properties": {k: v for k, v in event_props.items() if v is not None},
        }

        # Full URIs for relationship types so TripletStore stores hr:<type>
        # instead of the default urn:property:<type>, keeping SPARQL consistent.
        relationships.extend([
            # Shortcut edges — fast SPARQL when context is not needed
            {"source": person_id, "target": org_id,    "type": f"{BASE_URI}worksFor"},
            {"source": person_id, "target": role_id,   "type": f"{BASE_URI}hasRole"},
            # Reification spokes — full context via the event node
            {"source": event_id,  "target": person_id, "type": f"{BASE_URI}employee"},
            {"source": event_id,  "target": org_id,    "type": f"{BASE_URI}employer"},
            {"source": event_id,  "target": role_id,   "type": f"{BASE_URI}role"},
        ])

    return build_kg([{"entities": list(entities.values()), "relationships": relationships}])


kg = map_rows_to_kg(rows)
print(f"Entities built:      {len(kg.get('entities', []))}")
print(f"Relationships built: {len(kg.get('relationships', []))}")

sample = next((e for e in kg["entities"] if e["type"] == "EmploymentEvent"), None)
print(f"\nSample EmploymentEvent node: {sample}")

Step 5: Validate Ontology and Export OWL + SHACL

OntologyEngine validates your ontology dict and serialises it to standards-compliant files.

Output files:

• employment_manual_ontology.ttl — OWL 2 Turtle
• employment_manual_shapes.ttl — SHACL 1.1 node and property shapes

Standards status:

Standard	Semantica support
SPARQL 1.1	Full
SHACL 1.1 (sh:NodeShape, sh:PropertyShape, sh:minCount, sh:datatype, sh:class)	Full
SPARQL 1.2 (reifier annotation syntax, LATERAL)	Tracked — not yet implemented
SHACL 1.2 (sh:severity profiles, SHACL-AF extensions)	Tracked — not yet implemented

In [ ]:

engine = OntologyEngine(base_uri=BASE_URI)

validation = engine.validate(ontology)
owl_ttl    = engine.to_owl(ontology,   format="turtle")
shacl_ttl  = engine.to_shacl(ontology, format="turtle")

engine.export_owl(ontology,   "employment_manual_ontology.ttl", format="turtle")
engine.export_shacl(ontology, "employment_manual_shapes.ttl",   format="turtle")

print(f"Ontology valid:      {validation.valid}")
print(f"Ontology consistent: {validation.consistent}")
print(f"OWL output:          {len(owl_ttl):,} chars → employment_manual_ontology.ttl")
print(f"SHACL output:        {len(shacl_ttl):,} chars → employment_manual_shapes.ttl")

print("\n--- SHACL shapes (first 20 lines) ---")
print("\n".join(shacl_ttl.splitlines()[:20]))

Best-Practice Architecture

┌──────────────────────────────────┐
│  Ontology as code (Python dict)  │  ← versioned alongside your application
│  + AssociativeClass for n-ary    │
└───────────────┬──────────────────┘
                │ validate + export
                ▼
┌───────────────────────────────────┐
│  OWL 2 Turtle  │  SHACL 1.1      │  ← standards-compliant artifacts
└───────────────┬───────────────────┘
                │
                ▼
┌──────────────────────────────────┐
│  Snowflake — raw data access     │  ← no schema introspection
└───────────────┬──────────────────┘
                │ explicit mapping layer
                ▼
┌──────────────────────────────────┐
│  Ontology-aligned KG             │  ← types, IDs, edges match Step 1
└───────────────┬──────────────────┘
                │ optional
                ▼
┌──────────────────────────────────┐
│  Triplet store + SPARQL 1.1      │
└──────────────────────────────────┘

Why this split matters: If Semantica inferred the ontology from your Snowflake schema, every schema migration would risk silently changing your semantic model. With this pattern, schema changes only touch the mapping function in Step 4 — the ontology remains stable and under your control.

SPARQL Query Patterns

Two query styles are available because we wrote both shortcut edges and reification spokes.

Simple lookup — shortcut edge (no context needed)

PREFIX hr: <https://example.com/hr/>

SELECT ?personName ?orgName
WHERE {
    ?person  a hr:Person ;
             hr:name     ?personName ;
             hr:worksFor ?org .
    ?org     hr:legalName ?orgName .
}

Contextual lookup — via reification node (salary, dates, role)

PREFIX hr: <https://example.com/hr/>

SELECT ?personName ?roleTitle ?salary ?startDate
WHERE {
    ?event   a            hr:EmploymentEvent ;
             hr:employee  ?person ;
             hr:role      ?role ;
             hr:salary    ?salary ;
             hr:startDate ?startDate .
    ?person  hr:name   ?personName .
    ?role    hr:title  ?roleTitle .
}
ORDER BY DESC(?salary)

Future: SPARQL 1.2 reifier syntax

The SPARQL 1.2 draft introduces annotation syntax that lets you attach context directly to triples, without a separate intermediate node. Once the spec is ratified Semantica will adopt it, and the contextual query above may be expressible more concisely.

Step 6 (Optional): Load to Triplet Store and Run SPARQL

Set STORE_TO_TRIPLET=true to load the KG into a live triplet store and run the contextual reification query.

In [ ]:

if os.getenv("STORE_TO_TRIPLET", "false").lower() == "true":
    store = TripletStore(
        backend=os.getenv("TRIPLET_BACKEND", "blazegraph"),
        endpoint=os.getenv("TRIPLET_ENDPOINT", "http://localhost:9999/blazegraph"),
        namespace=os.getenv("TRIPLET_NAMESPACE", "kb"),
    )
    store_result = store.store(knowledge_graph=kg, ontology=ontology)
    print("Store result:", store_result)

    # Contextual reification query — person + role + salary via EmploymentEvent
    query = """
        PREFIX hr: <https://example.com/hr/>

        SELECT ?personName ?roleTitle ?salary ?startDate
        WHERE {
            ?event   a            hr:EmploymentEvent ;
                     hr:employee  ?person ;
                     hr:role      ?role ;
                     hr:salary    ?salary ;
                     hr:startDate ?startDate .
            ?person  hr:name   ?personName .
            ?role    hr:title  ?roleTitle .
        }
        ORDER BY DESC(?salary)
        LIMIT 10
    """
    result = store.execute_query(query)
    print(result)
else:
    print("Skipping triplet-store load/query (set STORE_TO_TRIPLET=true to enable)")