graflo.hq.rdf_inferencer

RDF / OWL ontology inference manager.

Reads the TBox (class & property definitions) from an RDF source and produces a graflo :class:Schema with vertices, edges, resources, and :class:Bindings.

The mapping follows these conventions:

• owl:Class / rdfs:Class -> Vertex
• owl:DatatypeProperty (rdfs:domain) -> Field on the domain vertex
• owl:ObjectProperty (rdfs:domain, rdfs:range) -> Edge (source = domain class, target = range class)
• Subject URI local name -> _key
• Inferred vertex identity is ["_uri"] so ABox ingestion keys on stable RDF subject URIs.
• owl:hasKey and inverse-functional datatype properties -> secondary identities (lookup keys) on the class; _uri stays the primary identity.
• Object properties (edges) become per-resource pipeline steps on the domain class that materialize the range vertex from the object-property field (URI reference) and emit the edge, including when domain and range are the same class (e.g. Publication -> Publication for cites). The subject and the objects of each property are kept in their own role, so an edge joins the subject to the objects of its own property and to nothing else. A property with several ranges reads each range from <property>@<Class>, the objects the source found typed with that class (typed_objects on the connector).

Classes, fields, edges and resources are ordered by IRI.

Requires rdflib (a core dependency of graflo).

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

RdfInferenceManager

Infer a graflo :class:Schema from an RDF / OWL ontology.

The manager reads the TBox (class and property declarations) from an rdflib Graph and constructs the corresponding graflo artefacts.

Attributes:

Name	Type	Description
target_db_flavor		Target graph-database flavour for downstream schema sanitisation.

Source code in graflo/hq/rdf_inferencer.py

class RdfInferenceManager:
    """Infer a graflo :class:`Schema` from an RDF / OWL ontology.

    The manager reads the TBox (class and property declarations) from an
    rdflib ``Graph`` and constructs the corresponding graflo artefacts.

    Attributes:
        target_db_flavor: Target graph-database flavour for downstream
            schema sanitisation.
    """

    def __init__(self, target_db_flavor: DBType = DBType.ARANGO):
        self.target_db_flavor = target_db_flavor

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def infer_schema(
        self,
        source: str | Path,
        *,
        endpoint_url: str | None = None,
        graph_uri: str | None = None,
        schema_name: str | None = None,
    ) -> tuple[Schema, IngestionModel]:
        """Infer a complete graflo Schema from an RDF/OWL ontology.

        Vertices use ``identity: ["_uri"]``. For each ``owl:ObjectProperty``, each
        source-class resource gets pipeline steps that (1) extract the range vertex
        from the predicate field via ``from: {_uri: <relation>}`` and (2) emit the
        edge, including when domain and range are the same class.

        Args:
            source: Path to an RDF file or a base URL (when using endpoint).
            endpoint_url: SPARQL endpoint to CONSTRUCT the ontology from.
            graph_uri: Named graph containing the ontology.
            schema_name: Name for the resulting schema.

        Returns:
            tuple[Schema, IngestionModel]: fully initialised schema and ingestion model.
        """
        from rdflib import OWL, RDF, RDFS

        g = _load_graph(source, endpoint_url=endpoint_url, graph_uri=graph_uri)

        # -- Discover classes -------------------------------------------------
        classes = _ontology_classes(g)  # local name -> IRI
        class_names = {uri: name for name, uri in classes.items()}

        logger.info("Discovered %d classes: %s", len(classes), list(classes.keys()))

        def named(prop: Any, predicate: Any) -> list[str]:
            """Vertex names of the classes *prop* points at, in IRI order."""
            return [
                class_names[uri]
                for uri in sorted({str(cls) for cls in g.objects(prop, predicate)})
                if uri in class_names
            ]

        # -- Discover datatype properties -> vertex fields --------------------
        fields_by_class: dict[str, list[str]] = {c: ["_key", "_uri"] for c in classes}

        for dp in sorted(set(g.subjects(RDF.type, OWL.DatatypeProperty)), key=str):
            dp_name = _local_name(str(dp))
            for domain_name in named(dp, RDFS.domain):
                if dp_name not in fields_by_class[domain_name]:
                    fields_by_class[domain_name].append(dp_name)

        # -- Discover object properties -> edges ------------------------------
        edges = _object_edges(g, classes)
        typed = _typed_properties(edges)

        logger.info("Discovered %d edges", len(edges))

        # -- Build Schema artefacts -------------------------------------------
        keys = _declared_keys(g, classes, fields_by_class)
        vertices = []
        for cls_name, fields in fields_by_class.items():
            vertex_fields = [VertexField(name=f) for f in fields]
            vertices.append(
                Vertex(
                    name=cls_name,
                    properties=vertex_fields,
                    identity=["_uri"],
                    secondary_identities=keys.get(cls_name, []),
                )
            )

        vertex_config = VertexConfig(vertices=vertices)

        inverses, symmetric = _declared_inverses(
            g, {e["relation"] for e in edges if e.get("relation")}
        )
        edge_objects = [
            Edge(
                source=e["source"],
                target=e["target"],
                relation=e.get("relation"),
                # A symmetric relation is realized by its edges being undirected.
                directed=e.get("relation") not in symmetric,
            )
            for e in edges
        ]
        edge_config = EdgeConfig(
            edges=edge_objects, inverses=inverses, symmetric=symmetric
        )
        logger.info(
            "Declared %d inverse pair(s) and %d symmetric relation(s)",
            len(inverses),
            len(symmetric),
        )

        # -- Build Resources (one per class) ----------------------------------
        edge_defs_by_source: dict[str, list[dict[str, str]]] = {}
        for edge_def in edges:
            edge_defs_by_source.setdefault(edge_def["source"], []).append(edge_def)

        resources: list[Resource] = []
        for cls_name in classes:
            # The object of a property is the target's `_uri`; a property with
            # several ranges reads each from the objects typed with it.
            references: list[Reference] = []
            for edge_def in edge_defs_by_source.get(cls_name, []):
                relation = field = edge_def["relation"]
                if relation in typed.get(cls_name, []):
                    field = f"{relation}@{edge_def['target']}"
                references.append(
                    (edge_def["target"], relation, {"_uri": field}, field)
                )
            resources.append(
                Resource(
                    name=cls_name,
                    pipeline=subject_with_references(cls_name, references),
                    infer_edges=not references,
                )
            )

        effective_name = schema_name or "rdf_schema"
        schema = Schema(
            metadata=GraphMetadata(name=effective_name),
            core_schema=CoreSchema(
                vertex_config=vertex_config, edge_config=edge_config
            ),
            db_profile=DatabaseProfile(db_flavor=self.target_db_flavor),
        )
        ingestion_model = IngestionModel(resources=resources)
        ingestion_model.finish_init(schema.core_schema)
        return schema, ingestion_model

    def create_bindings(
        self,
        source: str | Path,
        *,
        endpoint_url: str | None = None,
        graph_uri: str | None = None,
        data_file: str | Path | None = None,
    ) -> Bindings:
        """Create :class:`Bindings` from an RDF ontology.

        One :class:`SparqlConnector` is created per ``owl:Class`` / ``rdfs:Class``.
        The ontology is always loaded from *source* (a local file).  The
        *endpoint_url* is attached to each connector for runtime data queries
        but is **not** used to load the ontology itself.

        Args:
            source: Path to an RDF file containing the ontology.
            endpoint_url: SPARQL endpoint for the data (ABox) at runtime.
            graph_uri: Named graph containing the data.
            data_file: RDF file holding the data, when it is not *source*.

        Returns:
            Bindings with one SparqlConnector per class.
        """
        # Always load the ontology from the local file, not from the endpoint.
        g = _load_graph(source)
        classes = _ontology_classes(g)
        typed = _typed_properties(_object_edges(g, classes))

        bindings = Bindings()
        for cls_name, cls_uri in classes.items():
            connector = SparqlConnector(
                rdf_class=cls_uri,
                endpoint_url=endpoint_url,
                graph_uri=graph_uri,
                rdf_file=Path(data_file or source) if not endpoint_url else None,
                typed_objects=typed.get(cls_name, []),
            )
            bindings.add_connector(connector)
            bindings.bind_resource(cls_name, connector)

        logger.info(
            "Created %d SPARQL connectors from ontology",
            len(classes),
        )
        return bindings

Attributes

target_db_flavor = target_db_flavor instance-attribute

Methods:

__init__(target_db_flavor=DBType.ARANGO)

Source code in graflo/hq/rdf_inferencer.py

def __init__(self, target_db_flavor: DBType = DBType.ARANGO):
    self.target_db_flavor = target_db_flavor

create_bindings(source, *, endpoint_url=None, graph_uri=None, data_file=None)

Create :class:Bindings from an RDF ontology.

One :class:SparqlConnector is created per owl:Class / rdfs:Class. The ontology is always loaded from source (a local file). The endpoint_url is attached to each connector for runtime data queries but is not used to load the ontology itself.

Parameters:

Name	Type	Description	Default
source	str | Path	Path to an RDF file containing the ontology.	required
endpoint_url	str | None	SPARQL endpoint for the data (ABox) at runtime.	None
graph_uri	str | None	Named graph containing the data.	None
data_file	str | Path | None	RDF file holding the data, when it is not source.	None

Returns:

Type	Description
Bindings	Bindings with one SparqlConnector per class.

Source code in graflo/hq/rdf_inferencer.py

def create_bindings(
    self,
    source: str | Path,
    *,
    endpoint_url: str | None = None,
    graph_uri: str | None = None,
    data_file: str | Path | None = None,
) -> Bindings:
    """Create :class:`Bindings` from an RDF ontology.

    One :class:`SparqlConnector` is created per ``owl:Class`` / ``rdfs:Class``.
    The ontology is always loaded from *source* (a local file).  The
    *endpoint_url* is attached to each connector for runtime data queries
    but is **not** used to load the ontology itself.

    Args:
        source: Path to an RDF file containing the ontology.
        endpoint_url: SPARQL endpoint for the data (ABox) at runtime.
        graph_uri: Named graph containing the data.
        data_file: RDF file holding the data, when it is not *source*.

    Returns:
        Bindings with one SparqlConnector per class.
    """
    # Always load the ontology from the local file, not from the endpoint.
    g = _load_graph(source)
    classes = _ontology_classes(g)
    typed = _typed_properties(_object_edges(g, classes))

    bindings = Bindings()
    for cls_name, cls_uri in classes.items():
        connector = SparqlConnector(
            rdf_class=cls_uri,
            endpoint_url=endpoint_url,
            graph_uri=graph_uri,
            rdf_file=Path(data_file or source) if not endpoint_url else None,
            typed_objects=typed.get(cls_name, []),
        )
        bindings.add_connector(connector)
        bindings.bind_resource(cls_name, connector)

    logger.info(
        "Created %d SPARQL connectors from ontology",
        len(classes),
    )
    return bindings

infer_schema(source, *, endpoint_url=None, graph_uri=None, schema_name=None)

Infer a complete graflo Schema from an RDF/OWL ontology.

Vertices use identity: ["_uri"]. For each owl:ObjectProperty, each source-class resource gets pipeline steps that (1) extract the range vertex from the predicate field via from: {_uri: <relation>} and (2) emit the edge, including when domain and range are the same class.

Parameters:

Name	Type	Description	Default
source	str | Path	Path to an RDF file or a base URL (when using endpoint).	required
endpoint_url	str | None	SPARQL endpoint to CONSTRUCT the ontology from.	None
graph_uri	str | None	Named graph containing the ontology.	None
schema_name	str | None	Name for the resulting schema.	None

Returns:

Type	Description
tuple[Schema, IngestionModel]	tuple[Schema, IngestionModel]: fully initialised schema and ingestion model.

Source code in graflo/hq/rdf_inferencer.py

def infer_schema(
    self,
    source: str | Path,
    *,
    endpoint_url: str | None = None,
    graph_uri: str | None = None,
    schema_name: str | None = None,
) -> tuple[Schema, IngestionModel]:
    """Infer a complete graflo Schema from an RDF/OWL ontology.

    Vertices use ``identity: ["_uri"]``. For each ``owl:ObjectProperty``, each
    source-class resource gets pipeline steps that (1) extract the range vertex
    from the predicate field via ``from: {_uri: <relation>}`` and (2) emit the
    edge, including when domain and range are the same class.

    Args:
        source: Path to an RDF file or a base URL (when using endpoint).
        endpoint_url: SPARQL endpoint to CONSTRUCT the ontology from.
        graph_uri: Named graph containing the ontology.
        schema_name: Name for the resulting schema.

    Returns:
        tuple[Schema, IngestionModel]: fully initialised schema and ingestion model.
    """
    from rdflib import OWL, RDF, RDFS

    g = _load_graph(source, endpoint_url=endpoint_url, graph_uri=graph_uri)

    # -- Discover classes -------------------------------------------------
    classes = _ontology_classes(g)  # local name -> IRI
    class_names = {uri: name for name, uri in classes.items()}

    logger.info("Discovered %d classes: %s", len(classes), list(classes.keys()))

    def named(prop: Any, predicate: Any) -> list[str]:
        """Vertex names of the classes *prop* points at, in IRI order."""
        return [
            class_names[uri]
            for uri in sorted({str(cls) for cls in g.objects(prop, predicate)})
            if uri in class_names
        ]

    # -- Discover datatype properties -> vertex fields --------------------
    fields_by_class: dict[str, list[str]] = {c: ["_key", "_uri"] for c in classes}

    for dp in sorted(set(g.subjects(RDF.type, OWL.DatatypeProperty)), key=str):
        dp_name = _local_name(str(dp))
        for domain_name in named(dp, RDFS.domain):
            if dp_name not in fields_by_class[domain_name]:
                fields_by_class[domain_name].append(dp_name)

    # -- Discover object properties -> edges ------------------------------
    edges = _object_edges(g, classes)
    typed = _typed_properties(edges)

    logger.info("Discovered %d edges", len(edges))

    # -- Build Schema artefacts -------------------------------------------
    keys = _declared_keys(g, classes, fields_by_class)
    vertices = []
    for cls_name, fields in fields_by_class.items():
        vertex_fields = [VertexField(name=f) for f in fields]
        vertices.append(
            Vertex(
                name=cls_name,
                properties=vertex_fields,
                identity=["_uri"],
                secondary_identities=keys.get(cls_name, []),
            )
        )

    vertex_config = VertexConfig(vertices=vertices)

    inverses, symmetric = _declared_inverses(
        g, {e["relation"] for e in edges if e.get("relation")}
    )
    edge_objects = [
        Edge(
            source=e["source"],
            target=e["target"],
            relation=e.get("relation"),
            # A symmetric relation is realized by its edges being undirected.
            directed=e.get("relation") not in symmetric,
        )
        for e in edges
    ]
    edge_config = EdgeConfig(
        edges=edge_objects, inverses=inverses, symmetric=symmetric
    )
    logger.info(
        "Declared %d inverse pair(s) and %d symmetric relation(s)",
        len(inverses),
        len(symmetric),
    )

    # -- Build Resources (one per class) ----------------------------------
    edge_defs_by_source: dict[str, list[dict[str, str]]] = {}
    for edge_def in edges:
        edge_defs_by_source.setdefault(edge_def["source"], []).append(edge_def)

    resources: list[Resource] = []
    for cls_name in classes:
        # The object of a property is the target's `_uri`; a property with
        # several ranges reads each from the objects typed with it.
        references: list[Reference] = []
        for edge_def in edge_defs_by_source.get(cls_name, []):
            relation = field = edge_def["relation"]
            if relation in typed.get(cls_name, []):
                field = f"{relation}@{edge_def['target']}"
            references.append(
                (edge_def["target"], relation, {"_uri": field}, field)
            )
        resources.append(
            Resource(
                name=cls_name,
                pipeline=subject_with_references(cls_name, references),
                infer_edges=not references,
            )
        )

    effective_name = schema_name or "rdf_schema"
    schema = Schema(
        metadata=GraphMetadata(name=effective_name),
        core_schema=CoreSchema(
            vertex_config=vertex_config, edge_config=edge_config
        ),
        db_profile=DatabaseProfile(db_flavor=self.target_db_flavor),
    )
    ingestion_model = IngestionModel(resources=resources)
    ingestion_model.finish_init(schema.core_schema)
    return schema, ingestion_model

Functions: