Knowledge graph connector

Reading and writing the graph as if it were another connector.

Knowledge Graph Connector for KAPPS Semantic Middleware.

This module wraps kapps_ogm.OGM access behind the transitional_sync_middleware Connector protocol.

All middleware graph access goes through that one abstraction. Neither workflow registration code nor state registration code calls ogm.fetch or ogm.commit directly. One seam means the graph can be swapped, traced or stubbed in one place.

class kapps_semantic_middleware.connectors.knowledge_graph_connector.KnowledgeGraphConnector(ogm: OGM, instance_iri: IRI, *, class_scope: Any | None = None, class_spec: Any | None = None, materialize: bool = True, named_graph: str | None = None)

Bases: object

A Connector that provides synchronized access to a single entity in the knowledge graph.

This connector wraps OGM fetch/commit operations behind the transitional_sync_middleware Connector protocol, which enables uniform treatment of knowledge graph synchronization alongside OT device connectors (MQTT, OPC UA, HTTP).

async connect() → None

Establish a connection to the knowledge graph backend.

This is a lightweight liveness check. The OGM and GraphDB own their own connection lifecycle. We do not want to invent GraphDB methods that may not exist. This operation is a no-op. It returns immediately.

Raises:

ConnectionError – If the backend is not reachable (currently never raised).

async disconnect() → None

Disconnect from the knowledge graph backend.

This is a no-op since the OGM/GraphDB owns its own connection lifecycle.

Raises:

ConnectionError – If disconnection fails (currently never raised).

async provide() → Any

Fetch the current state of the graph entity.

Run OGM.fetch off the event loop via anyio.to_thread.run_sync to avoid blocking async execution.

Returns:

The fetched Node from the knowledge graph.

Raises:

ConnectionError – If fetching the data from the knowledge graph fails.

async consume(body: Any) → None

Commit updated data to the graph entity.

Run OGM.commit off the event loop via anyio.to_thread.run_sync to avoid blocking async execution.

Parameters:

body – The data dictionary to commit to the knowledge graph.

Raises:

ConnectionError – If committing the data to the knowledge graph fails.