graflo.hq.lookup

Which existing vertices a short text may name.

:func:lookup_candidates reads a graph for vertices a text mentions, by five lanes, all exact or rule-based -- no model and no vectors:

• exact: a run of the text's words equals a label field;
• key: a declared pattern matches the text and its value equals a key field;
• normalized: a run of words, through its source's normalizer, equals a derived normal-form field (derived: {form: normal});
• alias: a label lane finds a vertex of a declared alias type, and its alias edge leads to the target;
• compatible: a run of words, through its source's normalizer, is one of a derived keys field's less specific forms (derived: {form: keys}), read by ANY_IN.

It returns keys and labels only, never other fields, so a caller can show the candidates to a model without copying the graph's content into its prompt.

Attributes

LABEL_IRIS = frozenset({'http://www.w3.org/2004/02/skos/core#prefLabel', 'http://www.w3.org/2004/02/skos/core#altLabel', 'http://www.w3.org/2000/01/rdf-schema#label', 'https://schema.org/name', 'http://schema.org/name'}) module-attribute

Lane = Literal['exact', 'key', 'normalized', 'alias', 'compatible'] module-attribute

NOTATION_IRIS = frozenset({'http://www.w3.org/2004/02/skos/core#notation', 'http://purl.org/dc/terms/identifier', 'https://schema.org/identifier', 'http://schema.org/identifier'}) module-attribute

Readable = Callable[[str, str], bool] module-attribute

Classes

AliasHop

Bases: ConfigBaseModel

A vertex type whose labels name vertices of another type.

Attributes:

Name	Type	Description
alias_type	str	Type of the alias vertices (e.g. Alias).
relation	str	Edge relation from an alias to what it names.
target_type	str	Type the relation leads to.

Source code in graflo/hq/lookup.py

class AliasHop(ConfigBaseModel):
    """A vertex type whose labels name vertices of another type.

    Attributes:
        alias_type: Type of the alias vertices (e.g. ``Alias``).
        relation: Edge relation from an alias to what it names.
        target_type: Type the relation leads to.
    """

    alias_type: str
    relation: str
    target_type: str

Attributes

alias_type instance-attribute

relation instance-attribute

target_type instance-attribute

Candidate

Bases: ConfigBaseModel

A vertex a text may name.

Attributes:

Name	Type	Description
vertex_type	str	Logical type of the vertex.
key	dict[str, Any]	Its identity fields and values.
label	str | None	The label of its best label hit, if a label lane found it.
hits	list[LaneHit]	The best hit of each lane that found it, best first.

Source code in graflo/hq/lookup.py

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 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
    hits: list[LaneHit] = Field(min_length=1)

Attributes

hits = Field(min_length=1) class-attribute instance-attribute

key instance-attribute

label = None class-attribute instance-attribute

vertex_type instance-attribute

KeyPattern

Bases: ConfigBaseModel

A pattern whose match in a text is a key value of a vertex type.

Attributes:

Name	Type	Description
vertex_type	str	Logical vertex type the value identifies.
field	str	Logical field holding the value (an identity or secondary key).
pattern	str	Regular expression; its first group, else the whole match, is the value.

Source code in graflo/hq/lookup.py

class KeyPattern(ConfigBaseModel):
    """A pattern whose match in a text is a key value of a vertex type.

    Attributes:
        vertex_type: Logical vertex type the value identifies.
        field: Logical field holding the value (an identity or secondary key).
        pattern: Regular expression; its first group, else the whole match,
            is the value.
    """

    vertex_type: str
    field: str
    pattern: str

Attributes

field instance-attribute

pattern instance-attribute

vertex_type instance-attribute

LaneHit

Bases: ConfigBaseModel

One way a text names a vertex.

Attributes:

Name	Type	Description
lane	Lane	How it was found.
matched	str	The words or key value of the text it was found from.
targets	int | None	alias lane: how many vertices the alias reached under the guard; compatible lane: how many vertices of the type hold the matched form among their keys under the guard. One means the text names only this vertex in the reader's view.

Source code in graflo/hq/lookup.py

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.
        targets: ``alias`` lane: how many vertices the alias reached under the
            guard; ``compatible`` lane: how many vertices of the type hold the
            matched form among their keys under the guard. One means the text
            names only this vertex in the reader's view.
    """

    lane: Lane
    matched: str
    targets: int | None = None

Attributes

lane instance-attribute

matched instance-attribute

targets = None class-attribute instance-attribute

LookupSpec

Bases: ConfigBaseModel

What :func:lookup_candidates searches and how.

Attributes:

Name	Type	Description
vertex_types	list[str] | None	Types to return candidates of; default every type except the alias types.
label_fields	dict[str, list[str]]	Label fields per type, replacing those found by :func:label_fields.
key_patterns	list[KeyPattern]	Patterns for the key lane.
aliases	list[AliasHop]	Alias hops for the alias lane.
max_seeds	int	Most candidates returned.
max_words	int	Longest run of words tried as a label.

Source code in graflo/hq/lookup.py

class LookupSpec(ConfigBaseModel):
    """What :func:`lookup_candidates` searches and how.

    Attributes:
        vertex_types: Types to return candidates of; default every type except
            the alias types.
        label_fields: Label fields per type, replacing those found by
            :func:`label_fields`.
        key_patterns: Patterns for the ``key`` lane.
        aliases: Alias hops for the ``alias`` lane.
        max_seeds: Most candidates returned.
        max_words: Longest run of words tried as a label.
    """

    vertex_types: list[str] | None = None
    label_fields: dict[str, list[str]] = Field(default_factory=dict)
    key_patterns: list[KeyPattern] = Field(default_factory=list)
    aliases: list[AliasHop] = Field(default_factory=list)
    max_seeds: int = Field(default=20, ge=1)
    max_words: int = Field(default=4, ge=1)

Attributes

aliases = Field(default_factory=list) class-attribute instance-attribute

key_patterns = Field(default_factory=list) class-attribute instance-attribute

label_fields = Field(default_factory=dict) class-attribute instance-attribute

max_seeds = Field(default=20, ge=1) class-attribute instance-attribute

max_words = Field(default=4, ge=1) class-attribute instance-attribute

vertex_types = None class-attribute instance-attribute

Functions:

label_fields(vertex)

Fields of vertex whose semantics IRI is one of :data:LABEL_IRIS.

Source code in graflo/hq/lookup.py

def label_fields(vertex: Vertex) -> list[str]:
    """Fields of *vertex* whose semantics IRI is one of :data:`LABEL_IRIS`."""
    return [
        field.name
        for field in vertex.properties
        if field.semantics is not None and field.semantics.iri in LABEL_IRIS
    ]

lookup_candidates(db, schema, text, spec=None, *, guard=None, comparable=None)

Vertices of db that text may name, best first, at most max_seeds.

One candidate per vertex, carrying the best hit of every lane that found it, in the order exact, key, normalized, alias, compatible; within a lane the longer match wins. Vertices are ordered by their best hit.

Parameters:

Name	Type	Description	Default
db	Any	An open graph connection.	required
schema	Schema	The graph's logical schema.	required
text	str	The text to look in.	required
spec	LookupSpec | None	What to search; a default spec searches the label fields of every type.	None
guard	FilterExpression | None	A trusted read guard (see Connection.graph_neighbors); a vertex it hides is never a candidate, and an alias is followed only through elements it lets through.	None
comparable	Readable | None	Which fields may be matched on, as (vertex type, field) -> bool. A label, a key pattern's field or a derived field (or its source) it refuses is never read. Default: every field.	None

Returns:

Type	Description
list[Candidate]	The candidates; empty when the text names nothing found.

Raises:

Type	Description
NotImplementedError	A guard on a backend that cannot apply one, or a type with a derived keys field on a backend that cannot read one.

Source code in graflo/hq/lookup.py

def lookup_candidates(
    db: Any,
    schema: Schema,
    text: str,
    spec: LookupSpec | None = None,
    *,
    guard: FilterExpression | None = None,
    comparable: Readable | None = None,
) -> list[Candidate]:
    """Vertices of *db* that *text* may name, best first, at most ``max_seeds``.

    One candidate per vertex, carrying the best hit of every lane that found
    it, in the order ``exact``, ``key``, ``normalized``, ``alias``,
    ``compatible``; within a lane the longer match wins. Vertices are ordered
    by their best hit.

    Args:
        db: An open graph connection.
        schema: The graph's logical schema.
        text: The text to look in.
        spec: What to search; a default spec searches the label fields of
            every type.
        guard: A trusted read guard (see ``Connection.graph_neighbors``); a
            vertex it hides is never a candidate, and an alias is followed
            only through elements it lets through.
        comparable: Which fields may be matched on, as ``(vertex type, field)
            -> bool``. A label, a key pattern's field or a derived field (or
            its source) it refuses is never read. Default: every field.

    Returns:
        The candidates; empty when the text names nothing found.

    Raises:
        NotImplementedError: A *guard* on a backend that cannot apply one, or
            a type with a derived keys field on a backend that cannot read one.
    """
    spec = spec or LookupSpec()
    if guard is not None:
        db.require_read_guard()
    reader = _Reader(db, schema, guard)
    vertex_by_name = {v.name: v for v in schema.core_schema.vertex_config.vertices}
    alias_types = {hop.alias_type for hop in spec.aliases}
    types = spec.vertex_types or [
        name for name in vertex_by_name if name not in alias_types
    ]
    runs = word_runs(text, spec.max_words)
    readable: Readable = comparable or (lambda _type, _field: True)

    def labels_of(vertex_type: str) -> list[str]:
        if vertex_type in spec.label_fields:
            named = list(spec.label_fields[vertex_type])
        else:
            vertex = vertex_by_name.get(vertex_type)
            named = label_fields(vertex) if vertex is not None else []
        return [f for f in named if readable(vertex_type, f)]

    found: list[_Found] = []
    for vertex_type in types:
        vertex = vertex_by_name.get(vertex_type)
        labels = labels_of(vertex_type)
        found.extend(_label_lanes(reader, vertex, vertex_type, runs, labels, readable))
        found.extend(
            _compatible_lane(reader, vertex, vertex_type, runs, labels, readable)
        )

    for pattern in spec.key_patterns:
        if pattern.vertex_type not in types or not readable(
            pattern.vertex_type, pattern.field
        ):
            continue
        values = list(
            dict.fromkeys(
                match.group(1) if match.groups() else match.group(0)
                for match in re.finditer(pattern.pattern, text)
            )
        )
        identity = reader.identity(pattern.vertex_type)
        returned = list(dict.fromkeys([*identity, pattern.field]))
        for position, docs in reader.find(
            pattern.vertex_type, pattern.field, values, returned
        ).items():
            found.extend(
                _Found(
                    pattern.vertex_type,
                    {f: doc.get(f) for f in identity},
                    None,
                    LaneHit(lane="key", matched=values[position]),
                )
                for doc in docs
            )

    for hop in spec.aliases:
        if hop.target_type not in types:
            continue
        target_identity = reader.identity(hop.target_type)
        for alias in _label_lanes(
            reader,
            vertex_by_name.get(hop.alias_type),
            hop.alias_type,
            runs,
            labels_of(hop.alias_type),
            readable,
        ):
            reached = reader.neighbours(
                hop.alias_type, alias.key, hop.relation, hop.target_type
            )
            targets = {
                _identity_key({f: doc.get(f) for f in target_identity})
                for doc in reached
            }
            for doc in reached:
                found.append(
                    _Found(
                        hop.target_type,
                        {f: doc.get(f) for f in target_identity},
                        alias.label,
                        LaneHit(
                            lane="alias",
                            matched=alias.hit.matched,
                            targets=len(targets),
                        ),
                    )
                )

    return _merge(found)[: spec.max_seeds]

notation_fields(vertex)

Fields of vertex whose semantics IRI is one of :data:NOTATION_IRIS.

Source code in graflo/hq/lookup.py

def notation_fields(vertex: Vertex) -> list[str]:
    """Fields of *vertex* whose semantics IRI is one of :data:`NOTATION_IRIS`."""
    return [
        field.name
        for field in vertex.properties
        if field.semantics is not None and field.semantics.iri in NOTATION_IRIS
    ]

word_runs(text, max_words)

Runs of up to max_words words of text, longest first, without repeats.

Source code in graflo/hq/lookup.py

def word_runs(text: str, max_words: int) -> list[str]:
    """Runs of up to *max_words* words of *text*, longest first, without repeats."""
    words = [word.strip(_EDGE_PUNCTUATION) for word in text.split()]
    words = [word for word in words if word]
    runs: dict[str, None] = {}
    for size in range(min(max_words, len(words)), 0, -1):
        for start in range(len(words) - size + 1):
            runs.setdefault(" ".join(words[start : start + size]))
    return list(runs)