"""Lab 01 — a financial knowledge graph from scratch. Four pieces, and the reason there are four is the phase's whole argument: * an RDF TRIPLE STORE, because a bank's structural questions are about paths; * an RDFS/OWL REASONER, because "ultimately controlled by" is entailed, not stored; * a SHACL VALIDATOR, because OWL is open-world and cannot tell you a record is INCOMPLETE — which is the question a bank actually asks; * a SPARQL engine, because property paths express in one line what would otherwise be a recursive query nobody can read. pytest test_lab.py -v LAB_MODULE=solution pytest test_lab.py -v # the reference, must be green """ from __future__ import annotations import re from dataclasses import dataclass, field from enum import Enum from typing import Callable, Dict, FrozenSet, Iterable, List, Mapping, Optional, Sequence, Set, Tuple # ====================================================================================== # 1. IRIs, literals and triples # ====================================================================================== @dataclass(frozen=True, order=True) class IRI: """A globally unique identifier — the entire point of RDF.""" value: str def __str__(self) -> str: return self.value @dataclass(frozen=True, order=True) class Literal: """A value. Never a subject: you say things about entities, not about strings.""" value: str datatype: str = "xsd:string" def __str__(self) -> str: return self.value Term = object # IRI | Literal | Variable @dataclass(frozen=True, order=True) class Variable: name: str def __str__(self) -> str: return f"?{self.name}" @dataclass(frozen=True, order=True) class Triple: subject: IRI predicate: IRI object: Term def __str__(self) -> str: return f"{self.subject} {self.predicate} {self.object} ." DEFAULT_PREFIXES: Mapping[str, str] = { "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#", "rdfs": "http://www.w3.org/2000/01/rdf-schema#", "owl": "http://www.w3.org/2002/07/owl#", "sh": "http://www.w3.org/ns/shacl#", "xsd": "http://www.w3.org/2001/XMLSchema#", "fibo-be": "https://spec.edmcouncil.org/fibo/ontology/BE/LegalEntities/", "fibo-fnd": "https://spec.edmcouncil.org/fibo/ontology/FND/", "bank": "https://bank.example.ae/ontology/", "ent": "https://bank.example.ae/entity/", } _CURIE_RE = re.compile(r"^([A-Za-z][\w-]*):([^\s]+)$") class PrefixMap: """Expands CURIEs to full IRIs and shortens them back. A graph mixing expanded and abbreviated forms of the same IRI has two identifiers for one thing, which silently breaks every join. """ def __init__(self, prefixes: Optional[Mapping[str, str]] = None) -> None: self.prefixes: Dict[str, str] = dict(DEFAULT_PREFIXES) if prefixes: self.prefixes.update(prefixes) def expand(self, curie: str) -> IRI: """``bank:controls`` -> the full IRI. Unknown prefix -> KeyError. A FULL IRI is not a CURIE, even though ``https:...`` matches the CURIE shape — check for the scheme separator ``://`` first, or ``https://example.org/x`` is read as prefix "https". A bare name with no colon passes through unchanged. """ # TODO raise NotImplementedError def shorten(self, iri: IRI) -> str: """The inverse. When several prefixes match, prefer the LONGEST namespace.""" # TODO raise NotImplementedError # ====================================================================================== # 2. The triple store # ====================================================================================== class Graph: """An in-memory RDF graph with three indexes (SPO, POS, OSP). A triple store is a SET: asserting the same fact twice changes nothing. That idempotence is what makes entailment safe to run repeatedly. """ def __init__(self, prefixes: Optional[PrefixMap] = None) -> None: self.prefixes = prefixes or PrefixMap() self._triples: Set[Triple] = set() self._spo: Dict[IRI, Dict[IRI, Set[Term]]] = {} self._pos: Dict[IRI, Dict[Term, Set[IRI]]] = {} self._osp: Dict[Term, Dict[IRI, Set[IRI]]] = {} def add(self, subject: IRI, predicate: IRI, obj: Term) -> bool: """Insert into all three indexes. Return True only if the triple was NEW — that boolean is the signal a fixed-point loop needs.""" # TODO raise NotImplementedError def add_curies(self, subject: str, predicate: str, obj: str, *, literal: bool = False, datatype: str = "xsd:string") -> bool: """Convenience: expand CURIEs, wrapping ``obj`` in a Literal when asked.""" # TODO raise NotImplementedError def __len__(self) -> int: return len(self._triples) def __contains__(self, triple: Triple) -> bool: return triple in self._triples def triples(self) -> List[Triple]: """Sorted by (subject, predicate, object) so a dump is diffable.""" # TODO raise NotImplementedError def match(self, subject: Optional[IRI] = None, predicate: Optional[IRI] = None, obj: Optional[Term] = None) -> List[Triple]: """Pattern match with any position bound or free — the primitive SPARQL sits on. Pick the most selective index available (that is the only reason to keep three), then filter on the remaining bound positions. Return sorted output. """ # TODO raise NotImplementedError def objects(self, subject: IRI, predicate: IRI) -> List[Term]: """Sorted.""" # TODO raise NotImplementedError def subjects(self, predicate: IRI, obj: Term) -> List[IRI]: """Sorted.""" # TODO raise NotImplementedError # ====================================================================================== # 3. The reasoner # ====================================================================================== RDF_TYPE = IRI(DEFAULT_PREFIXES["rdf"] + "type") RDFS_SUBCLASS = IRI(DEFAULT_PREFIXES["rdfs"] + "subClassOf") RDFS_SUBPROPERTY = IRI(DEFAULT_PREFIXES["rdfs"] + "subPropertyOf") RDFS_DOMAIN = IRI(DEFAULT_PREFIXES["rdfs"] + "domain") RDFS_RANGE = IRI(DEFAULT_PREFIXES["rdfs"] + "range") OWL_INVERSE = IRI(DEFAULT_PREFIXES["owl"] + "inverseOf") OWL_TRANSITIVE = IRI(DEFAULT_PREFIXES["owl"] + "TransitiveProperty") OWL_SYMMETRIC = IRI(DEFAULT_PREFIXES["owl"] + "SymmetricProperty") @dataclass(frozen=True) class EntailmentReport: added: int rounds: int def materialize(graph: Graph, *, max_rounds: int = 50) -> EntailmentReport: """Forward-chain to a fixed point. Rules, with the names a reviewer will recognise: rdfs11 subClassOf is transitive rdfs9 type + subClassOf -> type rdfs5 subPropertyOf transitive rdfs7 assertion + subPropertyOf -> super-property assertion rdfs2 predicate + domain -> subject type rdfs3 predicate + range -> object type (IRI objects only — a literal is never a subject) owl inverseOf -> the reverse triple, BOTH directions owl TransitiveProperty -> composition owl SymmetricProperty -> the reverse triple Loop until a round adds nothing (or ``max_rounds``). Count rounds and total additions. Materialization is a CACHE, correct because RDFS/OWL entailment is MONOTONE: adding facts never retracts a conclusion. Deletion invalidates it. """ # TODO raise NotImplementedError # ====================================================================================== # 4. SHACL # ====================================================================================== class Severity(str, Enum): VIOLATION = "Violation" WARNING = "Warning" INFO = "Info" @dataclass(frozen=True) class PropertyShape: """The piece OWL cannot express: OWL's open-world assumption means "no LEI stated" is UNKNOWN, not missing.""" path: str # CURIE of the predicate min_count: Optional[int] = None max_count: Optional[int] = None datatype: Optional[str] = None pattern: Optional[str] = None node_kind: Optional[str] = None # "IRI" | "Literal" in_values: Optional[Tuple[str, ...]] = None severity: Severity = Severity.VIOLATION message: str = "" @dataclass(frozen=True) class NodeShape: name: str target_class: str # CURIE properties: Tuple[PropertyShape, ...] = () closed: bool = False # reject undeclared predicates ignored_properties: Tuple[str, ...] = () @dataclass(frozen=True) class Violation: focus_node: str path: str severity: Severity message: str value: Optional[str] = None def __str__(self) -> str: where = f"{self.focus_node} [{self.path}]" return f"{self.severity.value}: {where} {self.message}" + ( f" (value: {self.value})" if self.value else "") @dataclass(frozen=True) class ValidationReport: results: Tuple[Violation, ...] @property def conforms(self) -> bool: """VIOLATION fails; WARNING and INFO do not.""" # TODO raise NotImplementedError def by_severity(self, severity: Severity) -> List[Violation]: # TODO raise NotImplementedError def validate(graph: Graph, shapes: Sequence[NodeShape]) -> ValidationReport: """Validate every node targeted by each shape, sorted for determinism. Target selection reads the MATERIALIZED type, so a node typed only by entailment is still validated. That interaction is why materialization runs first. """ # TODO raise NotImplementedError def _validate_node(graph: Graph, focus: IRI, shape: NodeShape) -> List[Violation]: """Check every PropertyShape against this node. Order matters: cardinality first, then per-value checks. A failed node-kind or datatype check should CONTINUE to the next value rather than also reporting a pattern failure on a value of the wrong type. For ``closed=True``, report any predicate on the node that is neither declared by a property shape nor in ``ignored_properties`` (rdf:type is always allowed). """ # TODO raise NotImplementedError # ====================================================================================== # 5. SPARQL # ====================================================================================== Binding = Dict[str, Term] @dataclass(frozen=True) class PathStep: """``modifier``: "" (exactly one) · "+" (one or more) · "*" (zero or more). ``inverse`` traverses the predicate backwards — SPARQL's ``^``. Property paths are why SPARQL is worth implementing: "ultimately controlled by" is ``bank:controls+``, versus a recursive CTE nobody can read. """ predicate: str modifier: str = "" inverse: bool = False @dataclass(frozen=True) class Pattern: """Subject/object are a CURIE string, a Variable, or a Literal.""" subject: object path: PathStep object: object @dataclass(frozen=True) class Filter: fn: Callable[[Mapping[str, Term]], bool] description: str = "filter" @dataclass(frozen=True) class Optional_: """SPARQL OPTIONAL: a LEFT join. Non-matching patterns leave variables unbound rather than eliminating the solution.""" patterns: Tuple[Pattern, ...] @dataclass(frozen=True) class Query: select: Tuple[str, ...] where: Tuple[object, ...] # Pattern | Optional_ | Filter distinct: bool = True order_by: Tuple[str, ...] = () limit: Optional[int] = None def execute(graph: Graph, query: Query) -> List[Dict[str, str]]: """Evaluate by successive joins over a list of bindings. Start with ONE EMPTY binding. For each element: a Pattern extends every surviving binding with every way it can match; an Optional_ left-joins; a Filter removes bindings. **Shared variable names ARE the join condition** — there is no join clause because variable identity is the join. Then project: IRIs are shortened, literals stringified, unbound variables become "". Apply DISTINCT, sort by ``order_by`` (defaulting to ``select``), then ``limit``. """ # TODO raise NotImplementedError def _resolve(graph: Graph, term: object, binding: Binding) -> Optional[Term]: """A bound value for a term, or None when it is a free variable. A CURIE string expands; a Literal is itself.""" # TODO raise NotImplementedError def _join(graph: Graph, bindings: Sequence[Binding], pattern: Pattern) -> List[Binding]: """Extend each binding with every (subject, object) the pattern matches.""" # TODO raise NotImplementedError def _left_join(graph: Graph, bindings: Sequence[Binding], patterns: Sequence[Pattern]) -> List[Binding]: """Inner-join within the OPTIONAL block; if it yields nothing for a binding, keep the ORIGINAL binding unchanged.""" # TODO raise NotImplementedError def _walk(graph: Graph, step: PathStep, subject: Optional[Term], obj: Optional[Term]) -> List[Tuple[Term, Term]]: """Every (start, end) pair reachable by this path step, sorted. With no bound subject, the starts are every subject of the predicate (or every object when inverse). ``+`` and ``*`` are breadth-first WITH A VISITED SET — cycles in an ownership graph are real (circular shareholdings exist) and an unguarded traversal does not terminate. ``*`` includes the start itself. """ # TODO raise NotImplementedError # ====================================================================================== # 6. Graph-grounded retrieval # ====================================================================================== @dataclass(frozen=True) class GraphContext: entities: Tuple[str, ...] paths: Tuple[str, ...] rationale: str def expand_neighbourhood(graph: Graph, seed: str, *, hops: int = 1, predicates: Sequence[str] = ()) -> GraphContext: """Neighbourhood expansion — what a vector index cannot do. Breadth-first from ``seed`` for ``hops`` steps, following only ``predicates`` when given. Record, for each reached node, the path of ``a -pred-> b`` steps that got there. Return sorted entities (IRIs shortened, literals as text) and sorted unique path steps. ``hops < 0`` -> ValueError; ``hops == 0`` -> just the seed. """ # TODO raise NotImplementedError # ====================================================================================== # 7. The worked graph (used by the tests) # ====================================================================================== def build_ontology(graph: Graph) -> None: """A small FIBO-shaped ontology: faithful in structure, tiny in scope. Assert (via ``add_curies``): bank:Bank rdfs:subClassOf fibo-be:LegalEntity bank:Corporation rdfs:subClassOf fibo-be:LegalEntity fibo-be:LegalEntity rdfs:subClassOf fibo-fnd:AutonomousAgent bank:majorityOwns rdfs:subPropertyOf bank:controls bank:controls rdf:type owl:TransitiveProperty bank:controls owl:inverseOf bank:controlledBy bank:isCounterpartyOf rdf:type owl:SymmetricProperty bank:hasLEI rdfs:domain fibo-be:LegalEntity bank:majorityOwns rdfs:domain/rdfs:range fibo-be:LegalEntity bank:obligor rdfs:domain bank:Obligation ; rdfs:range fibo-be:LegalEntity """ # TODO raise NotImplementedError def build_facts(graph: Graph) -> None: """The instance data the tests and the worked example use. Ownership chain: Opaque -> Sanctioned -> Meridian -> Northgate -> Acme (Meridian bank:controlledBy Sanctioned; the rest via bank:majorityOwns.) Identifiers of varying quality — this is what makes SHACL interesting: ent:Acme hasLEI "5493001KJTIIGC8Y1R12" legalName "Acme Trading FZE" ent:Northgate hasLEI "NOT-AN-LEI" legalName "Northgate Holdings Ltd" ent:Meridian (no LEI) legalName "Meridian Capital" ent:Sanctioned hasLEI "5493009ABCDEFGH12345" legalName "Sanctioned Entity" ent:Opaque (no LEI, no legalName) -- the shell a KYC analyst cares about Types: Acme, Northgate, Meridian, Sanctioned are bank:Corporation. Also: ent:LoanA a bank:Obligation with bank:obligor ent:Acme and bank:amountAED "250000"^^xsd:integer; ent:Acme bank:isCounterpartyOf ent:Northgate; ent:Sanctioned bank:onSanctionsList "true"^^xsd:boolean. """ # TODO raise NotImplementedError LEGAL_ENTITY_SHAPE = NodeShape( name="LegalEntityShape", target_class="fibo-be:LegalEntity", properties=( PropertyShape(path="bank:hasLEI", min_count=1, max_count=1, datatype="xsd:string", pattern=r"^[A-Z0-9]{18}[0-9]{2}$", message="a legal entity needs exactly one well-formed LEI"), PropertyShape(path="bank:legalName", min_count=1, max_count=1, datatype="xsd:string"), ), ) OBLIGATION_SHAPE = NodeShape( name="ObligationShape", target_class="bank:Obligation", properties=( PropertyShape(path="bank:obligor", min_count=1, node_kind="IRI"), PropertyShape(path="bank:amountAED", min_count=1, max_count=1, datatype="xsd:integer"), ), ) def main() -> None: print("implement the TODOs, then compare with `python solution.py`") if __name__ == "__main__": main()