Skip to content

graflo.hq.rdf_landing

Land RDF facts on a manifest: read each class a resource handles from a file.

Facts extracted from documents arrive as an RDF file of instances (an ABox), often with the ontology they were extracted against in a second file. :func:land_rdf_facts turns them into manifest blocks:

  • A manifest with a schema and resources keeps both. Every class in the facts whose local name is a resource gets a :class:SparqlConnector over the facts file (re-landing the same class replaces its connector); the other classes are reported as unbound.
  • A manifest without them takes schema, resources and bindings inferred from the ontology (:class:RdfInferenceManager), every connector reading the facts file.

Attributes

__all__ = ['LandedClass', 'RdfLanding', 'fact_classes', 'land_rdf_facts'] module-attribute

Classes

LandedClass

Bases: ConfigBaseModel

A class found in the facts.

Source code in graflo/hq/rdf_landing.py 
45
46
47
48
49
50
class LandedClass(ConfigBaseModel):
    """A class found in the facts."""

    iri: str
    name: str = Field(description="Local name; the resource it lands on")
    instances: int

Attributes

instances instance-attribute
iri instance-attribute
name = Field(description='Local name; the resource it lands on') class-attribute instance-attribute

RdfLanding

Bases: ConfigBaseModel

The manifest with the facts landed, and what landed where.

Source code in graflo/hq/rdf_landing.py 
53
54
55
56
57
58
59
60
61
62
63
class RdfLanding(ConfigBaseModel):
    """The manifest with the facts landed, and what landed where."""

    manifest: GraphManifest
    bound: list[LandedClass] = Field(
        default_factory=list, description="Classes now read from the facts file"
    )
    unbound: list[LandedClass] = Field(
        default_factory=list,
        description="Classes in the facts that no resource of the manifest handles",
    )

Attributes

bound = Field(default_factory=list, description='Classes now read from the facts file') class-attribute instance-attribute
manifest instance-attribute
unbound = Field(default_factory=list, description='Classes in the facts that no resource of the manifest handles') class-attribute instance-attribute

Functions:

fact_classes(facts)

Domain classes instantiated in facts, with instance counts, in IRI order.

Source code in graflo/hq/rdf_landing.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def fact_classes(facts: Path) -> list[LandedClass]:
    """Domain classes instantiated in *facts*, with instance counts, in IRI order."""
    from rdflib import RDF, URIRef

    graph = _load_graph(facts)
    counts: dict[str, int] = {}
    for _, cls in graph.subject_objects(RDF.type):
        iri = str(cls)
        if isinstance(cls, URIRef) and not iri.startswith(_META_NAMESPACES):
            counts[iri] = counts.get(iri, 0) + 1
    return [
        LandedClass(iri=iri, name=_local_name(iri), instances=counts[iri])
        for iri in sorted(counts)
    ]

land_rdf_facts(manifest, facts, *, ontology=None, db_flavor=DBType.ARANGO)

Return manifest with facts bound to the resources that handle them.

Parameters:

Name Type Description Default
manifest GraphManifest | None

The manifest to land on; None or one without a schema and resources starts from the ontology.

 required
facts Path

RDF file of instances. Its path is written into the connectors as given, so pass the path the ingesting process will read.

 required
ontology Path | None

RDF file declaring the classes. Required when the manifest has no schema; otherwise used only to read multi-range properties by class (typed_objects).

 None
db_flavor DBType

Target flavour of a schema inferred from the ontology.

 ARANGO

Raises:

Type Description
ValueError

The manifest has no schema and no ontology was given.
Source code in graflo/hq/rdf_landing.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def land_rdf_facts(
    manifest: GraphManifest | None,
    facts: Path,
    *,
    ontology: Path | None = None,
    db_flavor: DBType = DBType.ARANGO,
) -> RdfLanding:
    """Return *manifest* with *facts* bound to the resources that handle them.

    Args:
        manifest: The manifest to land on; ``None`` or one without a schema and
            resources starts from the ontology.
        facts: RDF file of instances. Its path is written into the connectors
            as given, so pass the path the ingesting process will read.
        ontology: RDF file declaring the classes. Required when the manifest has
            no schema; otherwise used only to read multi-range properties by
            class (``typed_objects``).
        db_flavor: Target flavour of a schema inferred from the ontology.

    Raises:
        ValueError: The manifest has no schema and no ontology was given.
    """
    found = fact_classes(facts)
    if (
        manifest is None
        or manifest.graph_schema is None
        or manifest.ingestion_model is None
    ):
        if ontology is None:
            raise ValueError(
                "An ontology is required to land facts on a manifest without "
                "a schema and resources."
            )
        landed = _from_ontology(manifest, facts, ontology, db_flavor)
    else:
        landed = _onto_resources(manifest, facts, ontology)

    landed.finish_init()
    resources = {r.name for r in landed.require_ingestion_model().resources}
    return RdfLanding(
        manifest=landed,
        bound=[c for c in found if c.name in resources],
        unbound=[c for c in found if c.name not in resources],
    )