From f387c481b26ce0f4b01c6047dec1b7be8ec184ba Mon Sep 17 00:00:00 2001 From: Alexander Belikov Date: Sun, 25 Oct 2026 23:16:32 +0100 Subject: [PATCH] update lookup policy --- CHANGELOG.md | 17 ++ .../operations/finding_vertices_from_text.md | 19 +- docs/guides/database_connections.md | 4 + graflo/db/neo4j/conn.py | 12 +- graflo/hq/lookup.py | 104 ++++--- graflo/link/candidates.py | 276 ++++++++++++------ graflo/link/records.py | 74 ++++- pyproject.toml | 2 +- test/backends/test_neo4j_auth.py | 14 +- test/hq/test_lookup.py | 29 +- test/link/test_link_candidates.py | 162 +++++++++- uv.lock | 2 +- 12 files changed, 567 insertions(+), 148 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a85a9b9..8681bb6f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,23 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.24.0] + +### Changed + +- **Neo4j**: notifications about labels, properties and relation types nothing has written yet are + turned off at the driver. Neo4j 5.7 or later is required. +- **`lookup_candidates`** (breaking): `Candidate.lane` and `.mention` are replaced by `hits` + (`LaneHit {lane, matched}`), the best hit of every lane that found the vertex, best first. +- **`gather_candidates`** (`graflo.link`, breaking): + - `EntityCandidate.lane`, `.matched`, `.relation`, `.anchor` and `.role` are replaced by `lanes` + (`CandidateHit`, every lane that found the entity; each lane carries only its own fields). + - New `attributes` (`AttributeCheck {field, verdict, compare}`) check the record's other stated + properties against stored values, which are never returned. + - Key comparison is literal, as the field's declared type. + - New `comparable=` limits which stated fields are read; reserved fields never are. + - `max_candidates` counts entities after the key check. + ## [1.23.0] ### Added diff --git a/docs/concepts/operations/finding_vertices_from_text.md b/docs/concepts/operations/finding_vertices_from_text.md index 3fa7636a..65a4b05a 100644 --- a/docs/concepts/operations/finding_vertices_from_text.md +++ b/docs/concepts/operations/finding_vertices_from_text.md @@ -11,7 +11,7 @@ It uses no model and no vectors. A candidate is found in one of four ways: | `normalized` | A run of words, normalised, equals a stored normalised label field | | `alias` | A label lane finds an alias vertex, and its alias edge leads to the target | -Each candidate carries the vertex type, its identity fields, the label it was found by, the lane and the words of the text it came from — never the vertex's other fields. +Each candidate carries the vertex type, its identity fields, its label and `hits`: for every lane that found it, the words of the text it came from. It never carries the vertex's other fields. ## A first example @@ -28,8 +28,10 @@ spec = LookupSpec( candidates = lookup_candidates( conn, schema, "the Berlin pump is down, see SUP-42", spec ) -# [Candidate(vertex_type="Ticket", key={"number": "SUP-42"}, lane="key", ...), -# Candidate(vertex_type="Product", key={"sku": "P-200"}, label="Berlin pump", lane="alias", ...)] +# [Candidate(vertex_type="Ticket", key={"number": "SUP-42"}, +# hits=[LaneHit(lane="key", matched="SUP-42")]), +# Candidate(vertex_type="Product", key={"sku": "P-200"}, label="Berlin pump", +# hits=[LaneHit(lane="alias", matched="Berlin pump")])] ``` ## Label fields @@ -50,8 +52,8 @@ transforms: ## Results -- One candidate per vertex, from its best lane, in the order of the table above; within a lane, the longer mention first. -- At most `max_seeds` candidates (default 20). Runs of up to `max_words` words (default 4) are tried as labels. +- One candidate per vertex. Its `hits` hold the best hit of each lane that found it, in the order of the table above; within a lane, the longer match wins. Vertices are ordered by their best hit. +- At most `max_seeds` vertices (default 20). Runs of up to `max_words` words (default 4) are tried as labels. - `vertex_types` restricts the types returned. Alias types are not returned unless listed there. ## Candidates for staged records @@ -79,7 +81,10 @@ for entry in entries: ``` - A candidate never closes a record; only keys do. A pick, a rule or a person chooses. -- A candidate whose stored key or secondary-identity values contradict what the record states is left out. +- One candidate per entity. `lanes` lists every way it was found, best lane first. A `relation` hit names the relation and the anchor, a `metadata` hit (added by the caller) a role, and every other hit what matched. +- `attributes` checks each other property the record states against the entity's stored value: `agree`, `contradict`, or `unknown` when nothing is stored. Stored values are never returned. Equality is literal, as the field's declared type, so `"ANN@x.com"` is not `"ann@x.com"`. A field named in the spec's `normalized_fields` compares in `normalize_label` form (`compare: normalized`). +- A candidate whose stored key or secondary-identity values contradict what the record states is left out. Any other contradiction is reported, not a bar. +- `comparable(vertex_type, field)` limits which stated properties are probed, looked up by or checked, keys included. Reserved fields, such as access zones, are never compared. - A relation between two records states two entities: a record is never offered the entity a related record of the same source resolved to, and a relation whose two records resolve to one entity is not written. `promote` skips it and `resolution.dropped` lists it (`DroppedRelation`, reason `same_entity`); after fixing records by hand, `dropped_relations(resolution)` recomputes the list. -- Best lane first, at most `max_candidates` per record (default 10); `truncated` says when more were found. +- Best lane first, at most `max_candidates` entities per record (default 10), counted after lanes are merged and keys checked; `truncated` says when more were found. - `guard` limits every read, as in `resolve_records`. diff --git a/docs/guides/database_connections.md b/docs/guides/database_connections.md index 2fe52b4c..652a308a 100644 --- a/docs/guides/database_connections.md +++ b/docs/guides/database_connections.md @@ -38,6 +38,10 @@ PostgreSQL 15 or later, for the unique key on each edge table. An edge table created by an earlier version gains that key at schema definition or on its first ingest, which fails with a clear error if the table holds duplicate edges. +GraFlo needs Neo4j 5.7 or later. Its queries are built from the schema, so it +turns off Neo4j's notifications about labels, properties and relation types +that nothing has written yet. + A URI without a port gets the database's default. For Neo4j the default follows the scheme: `bolt_port` when it is set and otherwise 7687 for `bolt` and `neo4j` URIs, 7474 for `http` and `https`. diff --git a/graflo/db/neo4j/conn.py b/graflo/db/neo4j/conn.py index 5feeb4e3..7b2f5a13 100644 --- a/graflo/db/neo4j/conn.py +++ b/graflo/db/neo4j/conn.py @@ -27,7 +27,7 @@ from collections.abc import Sequence from typing import Any, LiteralString, cast -from neo4j import GraphDatabase +from neo4j import GraphDatabase, NotificationDisabledCategory from neo4j.exceptions import ClientError from graflo.architecture.graph_types import EdgeDirection, GraphContainer, Index @@ -173,7 +173,15 @@ def __init__(self, config: Neo4jConfig): auth = None if config.password is not None: auth = (config.username or DEFAULT_NEO4J_USERNAME, config.password) - self._driver = GraphDatabase.driver(uri=config.url, auth=auth) + # Queries are built from the schema, so an unrecognized label, property or + # relation is one nothing has written yet, never a typo: no notification. + self._driver = GraphDatabase.driver( + uri=config.url, + auth=auth, + notifications_disabled_categories=[ + NotificationDisabledCategory.UNRECOGNIZED + ], + ) self.conn = self._driver.session() def execute(self, query, **kwargs): diff --git a/graflo/hq/lookup.py b/graflo/hq/lookup.py index 1cdd8f6c..102547c8 100644 --- a/graflo/hq/lookup.py +++ b/graflo/hq/lookup.py @@ -17,7 +17,7 @@ from __future__ import annotations import re -from typing import Any, Literal +from typing import Any, Literal, NamedTuple from pydantic import Field @@ -100,22 +100,64 @@ class LookupSpec(ConfigBaseModel): max_words: int = Field(default=4, ge=1) +class LaneHit(ConfigBaseModel): + """One way a text names a vertex. + + Attributes: + lane: How it was found. + matched: The words or key value of the text it was found from. + """ + + lane: Lane + matched: str + + class Candidate(ConfigBaseModel): """A vertex a text may name. Attributes: vertex_type: Logical type of the vertex. key: Its identity fields and values. - label: The label it was found by, if a label lane found it. - lane: How it was found. - mention: The text it was found from. + label: The label of its best label hit, if a label lane found it. + hits: The best hit of each lane that found it, best first. """ vertex_type: str key: dict[str, Any] label: str | None = None - lane: Lane - mention: str + hits: list[LaneHit] = Field(min_length=1) + + +class _Found(NamedTuple): + vertex_type: str + key: dict[str, Any] + label: str | None + hit: LaneHit + + +def _rank(hit: LaneHit) -> tuple[int, int]: + return _LANE_RANK[hit.lane], -len(hit.matched) + + +def _merge(found: list[_Found]) -> list[Candidate]: + """One candidate per vertex with its best hit in each lane, best vertex first.""" + found = sorted(found, key=lambda f: _rank(f.hit)) + merged: dict[tuple[str, tuple[tuple[str, str], ...]], Candidate] = {} + for item in found: + identity_key = tuple(sorted((k, str(v)) for k, v in item.key.items())) + candidate = merged.get((item.vertex_type, identity_key)) + if candidate is None: + merged[(item.vertex_type, identity_key)] = Candidate( + vertex_type=item.vertex_type, + key=dict(item.key), + label=item.label, + hits=[item.hit], + ) + elif all(hit.lane != item.hit.lane for hit in candidate.hits): + candidate.hits.append(item.hit) + if candidate.label is None: + candidate.label = item.label + return list(merged.values()) def label_fields(vertex: Vertex) -> list[str]: @@ -212,12 +254,12 @@ def _label_lanes( spec: LookupSpec, runs: list[str], labels: list[str], -) -> list[Candidate]: +) -> list[_Found]: """The ``exact`` and ``normalized`` lanes for one type.""" identity = reader.identity(vertex_type) if not identity: return [] - found: list[Candidate] = [] + found: list[_Found] = [] normal = [normalize_label(run) for run in runs] lanes: list[tuple[Lane, list[str], list[str]]] = [ ("exact", labels, runs), @@ -236,15 +278,14 @@ def _label_lanes( ).items(): for doc in docs: found.append( - Candidate( - vertex_type=vertex_type, - key={f: doc.get(f) for f in identity}, - label=next( + _Found( + vertex_type, + {f: doc.get(f) for f in identity}, + next( (doc[f] for f in labels if doc.get(f) is not None), None, ), - lane=lane, - mention=mentions[position], + LaneHit(lane=lane, matched=mentions[position]), ) ) return found @@ -260,8 +301,9 @@ def lookup_candidates( ) -> list[Candidate]: """Vertices of *db* that *text* may name, best first, at most ``max_seeds``. - One candidate per vertex, from its best lane (``exact``, ``key``, - ``normalized``, ``alias``); within a lane, the longer mention first. + One candidate per vertex, carrying the best hit of every lane that found + it, in the order ``exact``, ``key``, ``normalized``, ``alias``; within a + lane the longer match wins. Vertices are ordered by their best hit. Args: db: An open graph connection. @@ -296,7 +338,7 @@ def labels_of(vertex_type: str) -> list[str]: vertex = vertex_by_name.get(vertex_type) return label_fields(vertex) if vertex is not None else [] - found: list[Candidate] = [] + found: list[_Found] = [] for vertex_type in types: found.extend( _label_lanes(reader, vertex_type, spec, runs, labels_of(vertex_type)) @@ -317,11 +359,11 @@ def labels_of(vertex_type: str) -> list[str]: pattern.vertex_type, pattern.field, values, returned ).items(): found.extend( - Candidate( - vertex_type=pattern.vertex_type, - key={f: doc.get(f) for f in identity}, - lane="key", - mention=values[position], + _Found( + pattern.vertex_type, + {f: doc.get(f) for f in identity}, + None, + LaneHit(lane="key", matched=values[position]), ) for doc in docs ) @@ -337,18 +379,12 @@ def labels_of(vertex_type: str) -> list[str]: hop.alias_type, alias.key, hop.relation, hop.target_type ): found.append( - Candidate( - vertex_type=hop.target_type, - key={f: doc.get(f) for f in target_identity}, - label=alias.label, - lane="alias", - mention=alias.mention, + _Found( + hop.target_type, + {f: doc.get(f) for f in target_identity}, + alias.label, + LaneHit(lane="alias", matched=alias.hit.matched), ) ) - found.sort(key=lambda c: (_LANE_RANK[c.lane], -len(c.mention))) - best: dict[tuple[str, tuple[tuple[str, str], ...]], Candidate] = {} - for candidate in found: - identity_key = tuple(sorted((k, str(v)) for k, v in candidate.key.items())) - best.setdefault((candidate.vertex_type, identity_key), candidate) - return list(best.values())[: spec.max_seeds] + return _merge(found)[: spec.max_seeds] diff --git a/graflo/link/candidates.py b/graflo/link/candidates.py index a3314033..4fe1b3c8 100644 --- a/graflo/link/candidates.py +++ b/graflo/link/candidates.py @@ -13,6 +13,12 @@ "the customer" that RAISED a resolved HD-4411 gives the customers linked to HD-4411 by ``RAISED``. +One candidate per entity carries every lane that found it, best first, and +the record's other stated properties checked against the entity's stored +values (``agree``, ``contradict``, ``unknown``); stored values are never +returned. Equality is literal, as the field's declared type; a declared +normalized field compares in ``normalize_label`` form. + Candidates never close a record: only keys do. A pick, a declared rule, an agent or a person chooses among them. A candidate whose stored key or secondary-identity values contradict what the record states is left out: a @@ -24,18 +30,23 @@ from __future__ import annotations from collections.abc import Callable -from typing import Any +from typing import Any, NamedTuple from graflo.architecture.graph_types.enums import EdgeDirection +from graflo.architecture.schema.coerce import CoercionError, coerce_value from graflo.architecture.schema.document import Schema from graflo.architecture.schema.key_contract import ( KeyContract, key_contract, key_status, ) +from graflo.architecture.schema.vertex import Field from graflo.filter.onto import FilterExpression from graflo.hq.lookup import LookupSpec, _Reader, label_fields, lookup_candidates from graflo.link.records import ( + AttributeCheck, + AttributeVerdict, + CandidateHit, CandidateLane, EntityCandidate, PartialRecord, @@ -44,11 +55,15 @@ StagedCandidates, StagedRecord, ) +from graflo.util.transform import normalize_label #: Words to look a staged record up by, besides its stated label values. LookupText = Callable[[PartialRecord], str | None] -#: Lanes best first; the cap keeps the best. +#: Whether a stated property ``(vertex type, field)`` may be read and compared. +Comparable = Callable[[str, str], bool] + +#: Lanes best first; candidates are ordered by their best lane. LANE_ORDER: dict[CandidateLane, int] = { "alternate": 0, "exact": 1, @@ -68,10 +83,12 @@ def _empty(value: Any) -> bool: return value is None or value == "" -def _same(a: Any, b: Any) -> bool: - if a == b: - return True - return str(a).strip().casefold() == str(b).strip().casefold() +class _Hit(NamedTuple): + """One lane's finding, before candidates are merged per entity.""" + + key: dict[str, Any] + label: str | None + hit: CandidateHit class _Gatherer: @@ -85,14 +102,21 @@ def __init__( guard: FilterExpression | None, spec: LookupSpec, lookup_text: LookupText | None, + comparable: Comparable | None, ) -> None: self.schema = schema self.db = db self.guard = guard self.spec = spec self.lookup_text = lookup_text + self.allowed = comparable + self.reserved = schema.reserved_fields() self.vertex_config = schema.core_schema.vertex_config self.vertices = {v.name: v for v in self.vertex_config.vertices} + self.fields: dict[str, dict[str, Field]] = { + name: {f.name: f for f in vertex.properties} + for name, vertex in self.vertices.items() + } self.contracts: dict[str, KeyContract] = {} self.records: dict[tuple[str, str], PartialRecord] = { (g.source_id, r.ref): r for g in resolution.graphs for r in g.records @@ -138,31 +162,70 @@ def label_of(self, vertex_type: str, doc: dict[str, Any]) -> str | None: return str(doc[name]) return None + # -- values -------------------------------------------------------------- + + def comparable(self, vertex_type: str, name: str) -> bool: + """A declared, unreserved field of the type that the caller lets through.""" + if name not in self.fields[vertex_type] or name in self.reserved: + return False + return self.allowed is None or self.allowed(vertex_type, name) + + def values(self, record: PartialRecord) -> dict[str, Any]: + """The record's stated values that may be read and compared.""" + return { + name: value + for name, value in record.values().items() + if not _empty(value) and self.comparable(record.type, name) + } + + def equal(self, vertex_type: str, name: str, a: Any, b: Any) -> bool: + """Whether *a* and *b* are one value of field *name*, as its declared type.""" + field = self.fields[vertex_type].get(name) + if field is not None: + try: + return coerce_value(a, field) == coerce_value(b, field) + except CoercionError: + pass + return a == b + + def freeze(self, vertex_type: str, key: dict[str, Any]) -> tuple: + """*key* as a hashable value, its fields read as their declared types.""" + fields = self.fields[vertex_type] + typed: dict[str, Any] = {} + for name, value in key.items(): + try: + typed[name] = coerce_value(value, fields[name]) + except (CoercionError, KeyError): + typed[name] = value + return _frozen(typed) + # -- lanes --------------------------------------------------------------- - def alternate(self, staged: StagedRecord) -> list[EntityCandidate]: + def alternate(self, staged: StagedRecord) -> list[_Hit]: """The entities a stated secondary identity matched, batch and graph.""" record = staged.record contract = self.contract(record.type) identity = list(contract.identity_fields) - values = record.values() - found: list[EntityCandidate] = [] + values = self.values(record) + found: list[_Hit] = [] for alternate in key_status(contract, values).alternates: probe = {f: values[f] for f in alternate.fields} - matched = ", ".join(f"{f}={values[f]}" for f in alternate.fields) + hit = CandidateHit( + lane="alternate", + matched=", ".join(f"{f}={values[f]}" for f in alternate.fields), + ) for key, item in self.resolved.items(): other = self.records.get(key) if item.type != record.type or other is None: continue stated = other.values() - if all(_same(stated.get(f), probe[f]) for f in alternate.fields): + if all( + self.equal(record.type, f, stated.get(f), probe[f]) + for f in alternate.fields + ): found.append( - EntityCandidate( - vertex_type=record.type, - key=dict(item.identity), - label=self.label_of(record.type, stated), - lane="alternate", - matched=matched, + _Hit( + dict(item.identity), self.label_of(record.type, stated), hit ) ) reader = self.reader_for(record.type) @@ -173,27 +236,25 @@ def alternate(self, staged: StagedRecord) -> list[EntityCandidate]: record.type, list(alternate.fields), [probe], returned ).get(0, []): found.append( - EntityCandidate( - vertex_type=record.type, - key={f: doc.get(f) for f in identity}, - label=self.label_of(record.type, doc), - lane="alternate", - matched=matched, + _Hit( + {f: doc.get(f) for f in identity}, + self.label_of(record.type, doc), + hit, ) ) return found - def lookup(self, staged: StagedRecord, limit: int) -> list[EntityCandidate]: + def lookup(self, staged: StagedRecord, limit: int) -> list[_Hit]: """``lookup_candidates`` over the stated labels and the lookup text.""" if self.db is None: return [] record = staged.record - values = record.values() + values = self.values(record) words = [self.lookup_text(record) or ""] if self.lookup_text else [] words += [ - str(values[f]) + values[f] for f in self.labels(record.type) - if not _empty(values.get(f)) and isinstance(values[f], str) + if isinstance(values.get(f), str) ] text = " ".join(dict.fromkeys(w.strip() for w in words if w and w.strip())) if not text: @@ -205,25 +266,24 @@ def lookup(self, staged: StagedRecord, limit: int) -> list[EntityCandidate]: self.db, self.schema, text, spec, guard=self.guard_for(record.type) ) return [ - EntityCandidate( - vertex_type=c.vertex_type, - key=dict(c.key), - label=c.label, - lane=c.lane, - matched=c.mention, + _Hit( + dict(c.key), + c.label, + CandidateHit(lane=hit.lane, matched=hit.matched), ) for c in found if c.vertex_type == record.type + for hit in c.hits ] - def relation(self, staged: StagedRecord) -> list[EntityCandidate]: + def relation(self, staged: StagedRecord) -> list[_Hit]: """Entities linked to a resolved anchor by a relation the text states.""" if self.db is None: return [] record = staged.record ref = record.ref identity = list(self.contract(record.type).identity_fields) - found: list[EntityCandidate] = [] + found: list[_Hit] = [] for held in self.held: relation = held.relation if held.source_id != staged.source_id: @@ -248,15 +308,11 @@ def relation(self, staged: StagedRecord) -> list[EntityCandidate]: guard=self.guard_for(record.type), ) for doc in reached.vertices.get(record.type, []): - key = {f: doc.get(f) for f in identity} found.append( - EntityCandidate( - vertex_type=record.type, - key=key, - label=self.label_of(record.type, doc), - lane="relation", - relation=name, - anchor=anchor_ref, + _Hit( + {f: doc.get(f) for f in identity}, + self.label_of(record.type, doc), + CandidateHit(lane="relation", relation=name, anchor=anchor_ref), ) ) return found @@ -288,45 +344,102 @@ def related(self, staged: StagedRecord) -> set[tuple[str, tuple]]: continue identity = self.contract(anchor.type).identity_fields key = {f: anchor.identity.get(f) for f in identity} - out.add((anchor.type, _frozen(key))) + out.add((anchor.type, self.freeze(anchor.type, key))) return out - def compatible( + def merge(self, staged: StagedRecord, hits: list[_Hit]) -> list[EntityCandidate]: + """One candidate per entity, with every lane that found it, best first.""" + vertex_type = staged.record.type + distinct = self.related(staged) + hits = [ + h + for h in hits + if not any(_empty(v) for v in h.key.values()) + and (vertex_type, self.freeze(vertex_type, h.key)) not in distinct + ] + hits.sort(key=lambda h: LANE_ORDER[h.hit.lane]) + merged: dict[tuple, EntityCandidate] = {} + for item in hits: + frozen = self.freeze(vertex_type, item.key) + candidate = merged.get(frozen) + if candidate is None: + merged[frozen] = EntityCandidate( + vertex_type=vertex_type, + key=dict(item.key), + label=item.label, + lanes=[item.hit], + ) + elif item.hit not in candidate.lanes: + candidate.lanes.append(item.hit) + if candidate.label is None: + candidate.label = item.label + return list(merged.values()) + + def check( self, staged: StagedRecord, candidates: list[EntityCandidate] ) -> list[EntityCandidate]: - """*candidates* whose stored keys agree with what the record states.""" + """*candidates* whose stored keys agree with the record, with its attributes. + + A contradicted key or secondary-identity field leaves the candidate out; + every other stated field the caller lets through is reported. + """ record = staged.record contract = self.contract(record.type) identity = list(contract.identity_fields) - values = {k: v for k, v in record.values().items() if not _empty(v)} + values = self.values(record) keyed = {f for key in contract.keys for f in key.fields} keyed |= {f for alt in contract.alternates for f in alt.fields} - checked = [f for f in values if f in keyed and f not in identity] + normalized = set(self.spec.normalized_fields.get(record.type, [])) kept = [ c for c in candidates - if all(_same(c.key.get(f), values[f]) for f in identity if f in values) + if all( + self.equal(record.type, f, c.key.get(f), values[f]) + for f in identity + if f in values + ) ] + compared = [f for f in values if f not in identity] reader = self.reader_for(record.type) - if not checked or not kept or reader is None: - return kept - stored = reader.find_by( - record.type, - identity, - [c.key for c in kept], - list(dict.fromkeys([*identity, *checked])), - ) + stored: dict[int, list[dict[str, Any]]] = {} + if compared and kept and reader is not None: + stored = reader.find_by( + record.type, + identity, + [c.key for c in kept], + list(dict.fromkeys([*identity, *compared])), + ) out = [] for position, candidate in enumerate(kept): docs = stored.get(position) - if not docs: # not in the graph (a batch entity): nothing to contradict - out.append(candidate) - continue - doc = docs[0] - if all(_empty(doc.get(f)) or _same(doc.get(f), values[f]) for f in checked): + doc = docs[0] if docs else {} # a batch entity: nothing stored + checks: list[AttributeCheck] = [] + for name in compared: + compare = "normalized" if name in normalized else "literal" + verdict = self.verdict( + record.type, name, compare, doc.get(name), values[name] + ) + if verdict == "contradict" and name in keyed: + break + checks.append( + AttributeCheck(field=name, verdict=verdict, compare=compare) + ) + else: + candidate.attributes = checks out.append(candidate) return out + def verdict( + self, vertex_type: str, name: str, compare: str, stored: Any, stated: Any + ) -> AttributeVerdict: + if _empty(stored): + return "unknown" + if compare == "normalized": + same = normalize_label(stored) == normalize_label(stated) + else: + same = self.equal(vertex_type, name, stored, stated) + return "agree" if same else "contradict" + def gather_candidates( schema: Schema, @@ -336,6 +449,7 @@ def gather_candidates( guard: FilterExpression | None = None, spec: LookupSpec | None = None, lookup_text: LookupText | None = None, + comparable: Comparable | None = None, max_candidates: int = 10, ) -> list[StagedCandidates]: """The entities each staged record of *resolution* may be, best first. @@ -346,7 +460,8 @@ def gather_candidates( returned. Relation-path anchors are its resolved records, so run this after their write when they are new to the graph. db: An open connection. Without one, only the batch's resolved records - are searched (the ``alternate`` lane). + are searched (the ``alternate`` lane), and every attribute is + ``unknown``. guard: A trusted read guard every graph read passes, as in ``resolve_records``: it applies to the reads of a type that declares every field it names, and on a relation path to every @@ -354,12 +469,18 @@ def gather_candidates( guards. spec: Lookup spec for the label lanes (normalised fields, key patterns, alias hops, label field overrides). Its ``vertex_types`` - and ``max_seeds`` are set per record. + and ``max_seeds`` are set per record; its normalised fields are + compared in ``normalize_label`` form. lookup_text: Words to look each staged record up by besides the label values it states, such as the passage it was read from; typically taken from the caller's own ``evidence``. - max_candidates: Most candidates per record; ``truncated`` says when - more were found. + comparable: Which stated properties may be read and compared, as + ``(vertex type, field) -> bool``; one it refuses is never probed, + looked up by or checked, keys included. Declared, unreserved + fields are the only ones ever considered. + max_candidates: Most candidates per record, counted as entities after + lanes are merged and keys checked; ``truncated`` says when more + were found. Returns: One entry per staged record, in the resolution's order; an entry with @@ -372,9 +493,9 @@ def gather_candidates( raise ValueError("max_candidates must be at least 1") if guard is not None and db is not None: db.require_read_guard() - gatherer = _Gatherer( - schema, resolution, db, guard, spec or LookupSpec(), lookup_text - ) + spec = spec or LookupSpec() + gatherer = _Gatherer(schema, resolution, db, guard, spec, lookup_text, comparable) + limit = max(spec.max_seeds, max_candidates + 1) out: list[StagedCandidates] = [] for staged in resolution.staged.records: record = staged.record @@ -386,23 +507,12 @@ def gather_candidates( out.append(entry) if not gatherer.contract(record.type).identity_fields: continue - found = [ + hits = [ *gatherer.alternate(staged), - *gatherer.lookup(staged, max_candidates + 1), + *gatherer.lookup(staged, limit), *gatherer.relation(staged), ] - distinct = gatherer.related(staged) - found = [ - c - for c in found - if not any(_empty(v) for v in c.key.values()) - and (c.vertex_type, _frozen(c.key)) not in distinct - ] - found.sort(key=lambda c: LANE_ORDER[c.lane]) - best: dict[tuple[tuple[str, str], ...], EntityCandidate] = {} - for candidate in found: - best.setdefault(_frozen(candidate.key), candidate) - kept = gatherer.compatible(staged, list(best.values())) + kept = gatherer.check(staged, gatherer.merge(staged, hits)) entry.candidates = kept[:max_candidates] entry.truncated = len(kept) > max_candidates return out diff --git a/graflo/link/records.py b/graflo/link/records.py index 69c7771a..0fa8ad66 100644 --- a/graflo/link/records.py +++ b/graflo/link/records.py @@ -219,12 +219,13 @@ def identity_of(self, source_id: str, ref: str) -> dict[str, Any] | None: ] -class EntityCandidate(ConfigBaseModel): - """An entity a staged record may be. A candidate never closes a record.""" +class CandidateHit(ConfigBaseModel): + """One way an entity was found for a staged record. + + Each lane carries its own fields and no others: ``relation`` the relation + and the anchor, ``metadata`` a role, every other lane what matched. + """ - vertex_type: str - key: dict[str, Any] = Field(description="Its identity fields and values.") - label: str | None = Field(default=None, description="Its label, if it has one.") lane: CandidateLane = Field( description=( "How it was found. ``alternate``: one of the entities a stated " @@ -251,6 +252,69 @@ class EntityCandidate(ConfigBaseModel): description="``metadata`` lane: a role the caller assigns.", ) + @model_validator(mode="after") + def _fields_of_its_lane(self) -> CandidateHit: + if self.lane == "relation": + expected = {"relation", "anchor"} + elif self.lane == "metadata": + expected = {"role"} + else: + expected = {"matched"} + stated = { + name + for name in ("matched", "relation", "anchor", "role") + if getattr(self, name) is not None + } + if stated != expected: + raise ValueError( + f"a {self.lane} hit carries {sorted(expected)}, got {sorted(stated)}" + ) + return self + + +#: How a stated value was compared with the stored one. +AttributeVerdict = Literal["agree", "contradict", "unknown"] + + +class AttributeCheck(ConfigBaseModel): + """A stated property of the record against the candidate's stored value. + + The stored value itself is never carried. + """ + + field: str + verdict: AttributeVerdict = Field( + description=( + "``agree`` / ``contradict``: the stated and stored values are " + "equal / differ. ``unknown``: nothing is stored to compare with." + ) + ) + compare: Literal["literal", "normalized"] = Field( + description=( + "``literal``: equal as the field's declared type. ``normalized``: " + "equal in ``normalize_label`` form, for a declared normalized field." + ) + ) + + +class EntityCandidate(ConfigBaseModel): + """An entity a staged record may be. A candidate never closes a record.""" + + vertex_type: str + key: dict[str, Any] = Field(description="Its identity fields and values.") + label: str | None = Field(default=None, description="Its label, if it has one.") + lanes: list[CandidateHit] = Field( + min_length=1, description="Every way it was found, best first." + ) + attributes: list[AttributeCheck] = Field( + default_factory=list, + description=( + "The record's other comparable stated properties against the " + "candidate's stored values. A contradicted key never appears: it " + "leaves the candidate out." + ), + ) + class StagedCandidates(ConfigBaseModel): """The candidates of one staged record, best first.""" diff --git a/pyproject.toml b/pyproject.toml index 7a4e2875..2bc8f806 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,7 +45,7 @@ license-files = ["LICENSE", "NOTICE"] name = "graflo" readme = "README.md" requires-python = ">=3.11" -version = "1.23.0" +version = "1.24.0" [project.optional-dependencies] dev = [ diff --git a/test/backends/test_neo4j_auth.py b/test/backends/test_neo4j_auth.py index fd24bd64..fd973122 100644 --- a/test/backends/test_neo4j_auth.py +++ b/test/backends/test_neo4j_auth.py @@ -1,9 +1,11 @@ -"""Neo4j credentials: a password is never dropped.""" +"""Neo4j driver options: a password is never dropped; unknown-schema notifications are off.""" from __future__ import annotations from unittest.mock import patch +from neo4j import NotificationDisabledCategory + from graflo.connections.onto import Neo4jConfig from graflo.db.neo4j.conn import Neo4jConnection @@ -18,3 +20,13 @@ def test_a_password_without_a_username_authenticates_as_neo4j(mock_driver): def test_no_password_connects_unauthenticated(mock_driver): Neo4jConnection(Neo4jConfig(uri="bolt://localhost:7687", username="neo4j")) assert mock_driver.call_args.kwargs["auth"] is None + + +@patch("graflo.db.neo4j.conn.GraphDatabase.driver") +def test_unrecognized_schema_notifications_are_disabled(mock_driver): + Neo4jConnection(Neo4jConfig(uri="bolt://localhost:7687", password="secret")) + kwargs = mock_driver.call_args.kwargs + assert kwargs["notifications_disabled_categories"] == [ + NotificationDisabledCategory.UNRECOGNIZED + ] + assert "notifications_disabled_classifications" not in kwargs diff --git a/test/hq/test_lookup.py b/test/hq/test_lookup.py index 20596ea7..c12440ee 100644 --- a/test/hq/test_lookup.py +++ b/test/hq/test_lookup.py @@ -19,6 +19,7 @@ from graflo.filter.onto import ComparisonOperator, FilterExpression from graflo.hq.lookup import ( AliasHop, + Candidate, KeyPattern, LookupSpec, label_fields, @@ -121,7 +122,11 @@ def conn(tmp_path: Path) -> Iterator[Any]: def _found(candidates: list[Any]) -> set[tuple[str, str, str]]: - return {(c.vertex_type, next(iter(c.key.values())), c.lane) for c in candidates} + return { + (c.vertex_type, next(iter(c.key.values())), hit.lane) + for c in candidates + for hit in c.hits + } class TestNormalizeLabel: @@ -171,22 +176,32 @@ def test_an_alias_hops_to_its_target(self, conn: Any) -> None: conn, _schema(), "the berlin pump is down again", SPEC ) targets = [c for c in found if c.vertex_type == "Product"] - assert [(c.key, c.lane) for c in targets] == [({"sku": "P-200"}, "alias")] - assert targets[0].mention.lower() == "berlin pump" + assert [(c.key, [h.lane for h in c.hits]) for c in targets] == [ + ({"sku": "P-200"}, ["alias"]) + ] + assert targets[0].hits[0].matched.lower() == "berlin pump" - def test_a_candidate_carries_its_label_and_mention(self, conn: Any) -> None: + def test_a_candidate_carries_its_label_and_what_matched(self, conn: Any) -> None: (candidate,) = [ c for c in lookup_candidates(conn, _schema(), "Valve Pro leaks", SPEC) if c.vertex_type == "Product" ] assert candidate.label == "Valve Pro" - assert candidate.mention == "Valve Pro" + assert candidate.hits[0].matched == "Valve Pro" - def test_one_candidate_per_vertex_best_lane_first(self, conn: Any) -> None: + def test_one_candidate_per_vertex_with_every_lane_best_first( + self, conn: Any + ) -> None: found = lookup_candidates(conn, _schema(), "Pump X1 or pump x1", SPEC) products = [c for c in found if c.vertex_type == "Product"] - assert [(c.key["sku"], c.lane) for c in products] == [("P-200", "exact")] + assert [(c.key["sku"], [h.lane for h in c.hits]) for c in products] == [ + ("P-200", ["exact", "normalized"]) + ] + + def test_a_candidate_has_at_least_one_hit(self) -> None: + with pytest.raises(ValueError, match="hits"): + Candidate(vertex_type="Product", key={"sku": "P-200"}, hits=[]) def test_the_cap_holds(self, conn: Any) -> None: spec = SPEC.model_copy(update={"max_seeds": 1}) diff --git a/test/link/test_link_candidates.py b/test/link/test_link_candidates.py index ad5dd093..2aab3fbf 100644 --- a/test/link/test_link_candidates.py +++ b/test/link/test_link_candidates.py @@ -19,6 +19,8 @@ from graflo.hq.lookup import AliasHop, KeyPattern, LookupSpec from graflo.link.candidates import gather_candidates from graflo.link.records import ( + CandidateHit, + EntityCandidate, PartialRecord, RecordGraph, RecordRelation, @@ -51,7 +53,11 @@ }, { "name": "Product", - "properties": ["sku", {"name": "name", "semantics": NAME}], + "properties": [ + "sku", + {"name": "name", "semantics": NAME}, + "name_norm", + ], "identity": ["sku"], }, { @@ -137,7 +143,9 @@ def _seed(conf: GraFloBackendConfig) -> None: {"number": "HD-4411", "title": "Pump leak"}, {"number": "HD-9"}, ], - "Product": [{"sku": "P-200", "name": "Pump X1"}], + "Product": [ + {"sku": "P-200", "name": "Pump X1", "name_norm": "pump x1"} + ], "Alias": [{"label": "Berlin pump"}], }, edges={ @@ -171,15 +179,26 @@ def _gather(db: Any, graphs: list[RecordGraph], **kwargs: Any): def _found(entry: StagedCandidates) -> list[tuple[str, str]]: - return [(next(iter(c.key.values())), c.lane) for c in entry.candidates] + """Each candidate's first key value and its best lane.""" + return [(next(iter(c.key.values())), c.lanes[0].lane) for c in entry.candidates] + + +def _lanes(candidate: EntityCandidate) -> list[str]: + return [hit.lane for hit in candidate.lanes] + + +def _attributes(candidate: EntityCandidate) -> list[tuple[str, str, str]]: + return [(a.field, a.verdict, a.compare) for a in candidate.attributes] def test_a_stated_label_finds_its_entity_exactly(db): (entry,) = _gather(db, [_g("m1", [_m("c", "Customer", "Ann Lee", name="Ann Lee")])]) assert (entry.record.ref, entry.reason) == ("c", "no_key") assert _found(entry) == [("ACC-2", "exact")] - assert entry.candidates[0].label == "Ann Lee" - assert entry.candidates[0].matched == "Ann Lee" + (candidate,) = entry.candidates + assert candidate.label == "Ann Lee" + assert candidate.lanes[0].matched == "Ann Lee" + assert _attributes(candidate) == [("name", "agree", "literal")] def test_an_alias_leads_to_what_it_names(db): @@ -196,7 +215,8 @@ def test_a_stated_relation_to_a_resolved_anchor_gives_the_linked_entities(db): (entry,) = _gather(db, [graph]) assert _found(entry) == [("ACC-2", "relation")] (candidate,) = entry.candidates - assert (candidate.relation, candidate.anchor) == ("RAISED", "t") + (hit,) = candidate.lanes + assert (hit.relation, hit.anchor, hit.matched) == ("RAISED", "t", None) assert candidate.label == "Ann Lee" @@ -238,7 +258,7 @@ def test_a_candidate_contradicting_a_stated_key_is_left_out(db): assert _found(entry) == [("ACC-6", "exact")] -def test_one_candidate_per_entity_from_its_best_lane(db): +def test_one_candidate_per_entity_with_every_lane_best_first(db): graph = _g( "m1", [ @@ -248,7 +268,135 @@ def test_one_candidate_per_entity_from_its_best_lane(db): [(RAISED, "c", "t")], ) (entry,) = _gather(db, [graph]) + (candidate,) = entry.candidates + assert candidate.key == {"account": "ACC-2"} + assert _lanes(candidate) == ["exact", "relation"] + + +def test_the_cap_counts_entities_not_lanes(db): + graph = _g( + "m1", + [ + _m("c", "Customer", "Ann Lee", name="Ann Lee"), + _m("t", "Ticket", number="HD-4411"), + ], + [(RAISED, "c", "t")], + ) + (entry,) = _gather(db, [graph], max_candidates=1) assert _found(entry) == [("ACC-2", "exact")] + assert not entry.truncated + + +def test_a_dropped_candidate_never_takes_the_place_of_a_kept_one(db): + # Both desks and ACC-1 hold another email; only ACC-6 can be this record, + # though the desks are found first by their longer names. + graph = _g( + "m1", + [ + _m( + "c", + "Customer", + "Front Desk and Back Desk and Jo Park", + email="jo2@x.com", + ) + ], + ) + (entry,) = _gather(db, [graph], max_candidates=1) + assert _found(entry) == [("ACC-6", "exact")] + assert not entry.truncated + + +def test_a_non_key_contradiction_is_shown_not_dropped(db): + graph = _g("m1", [_m("p", "Product", "Pump X1", name="Pump X2")]) + (entry,) = _gather(db, [graph]) + (candidate,) = entry.candidates + assert candidate.key == {"sku": "P-200"} + assert _attributes(candidate) == [("name", "contradict", "literal")] + + +def test_nothing_stored_is_unknown(db): + # ACC-2 holds no email: the stated one neither agrees nor contradicts. + graph = _g( + "m1", [_m("c", "Customer", "Ann Lee", name="Ann Lee", email="ann@lee.com")] + ) + (entry,) = _gather(db, [graph]) + (candidate,) = entry.candidates + assert candidate.key == {"account": "ACC-2"} + assert _attributes(candidate) == [ + ("name", "agree", "literal"), + ("email", "unknown", "literal"), + ] + + +def test_a_key_differing_only_in_case_contradicts(db): + graph = _g("m1", [_m("c", "Customer", "Jo Park", name="Jo Park", email="JO@x.com")]) + (entry,) = _gather(db, [graph]) + assert _found(entry) == [("ACC-6", "exact")] + + +def test_a_batch_alternate_matches_literally(): + graphs = [ + _g("a", [_m("c", "Customer", account="ACC-8", email="x@x.com")]), + _g("b", [_m("c", "Customer", account="ACC-9", email="x@x.com")]), + _g("d", [_m("c", "Customer", account="ACC-7", email="X@x.com")]), + _g("c", [_m("c", "Customer", email="x@x.com")]), + ] + schema = _schema() + resolution = resolve_records(schema, graphs) + (entry,) = gather_candidates(schema, resolution) + assert sorted(_found(entry)) == [("ACC-8", "alternate"), ("ACC-9", "alternate")] + assert all( + _attributes(c) == [("email", "unknown", "literal")] for c in entry.candidates + ) + + +def test_a_normalized_field_compares_in_normal_form(db): + spec = SPEC.model_copy(update={"normalized_fields": {"Product": ["name_norm"]}}) + graph = _g( + "m1", [_m("p", "Product", "Pump X1", name="Pump X1", name_norm="PUMP-x1")] + ) + (entry,) = _gather(db, [graph], spec=spec) + (candidate,) = entry.candidates + assert _attributes(candidate) == [ + ("name", "agree", "literal"), + ("name_norm", "agree", "normalized"), + ] + + +def test_a_field_the_caller_refuses_is_never_compared(db): + # Compared, email leaves ACC-1 out; refused, it neither drops nor shows. + graph = _g("m1", [_m("c", "Customer", "Jo Park", name="Jo Park", email="zz@x.com")]) + (entry,) = _gather(db, [graph], comparable=lambda _type, field: field != "email") + assert sorted(_found(entry)) == [("ACC-1", "exact"), ("ACC-6", "exact")] + assert all( + _attributes(c) == [("name", "agree", "literal")] for c in entry.candidates + ) + + +def test_a_reserved_field_is_never_compared(zoned): + graph = _g("m1", [_m("c", "Customer", "Ann Lee", name="Ann Lee", access_zones=[2])]) + (entry,) = _gather(zoned, [graph], schema=_zoned_schema()) + (candidate,) = entry.candidates + assert _attributes(candidate) == [("name", "agree", "literal")] + + +@pytest.mark.parametrize( + ("fields", "message"), + [ + ({"lane": "relation", "relation": "RAISED"}, "anchor"), + ({"lane": "exact"}, "matched"), + ({"lane": "exact", "matched": "Ann", "role": "author"}, "role"), + ({"lane": "metadata", "matched": "Ann"}, "role"), + ], +) +def test_a_hit_carries_exactly_the_fields_of_its_lane(fields, message): + with pytest.raises(ValueError, match=message): + CandidateHit(**fields) + + +def test_a_candidate_has_at_least_one_lane(): + with pytest.raises(ValueError, match="lanes"): + EntityCandidate(vertex_type="Customer", key={"account": "ACC-2"}, lanes=[]) def test_the_cap_holds_and_says_so(db): diff --git a/uv.lock b/uv.lock index 7b4688be..6a8f0556 100644 --- a/uv.lock +++ b/uv.lock @@ -568,7 +568,7 @@ wheels = [ [[package]] name = "graflo" -version = "1.23.0" +version = "1.24.0" source = { editable = "." } dependencies = [ { name = "boto3" },