Skip to content
HyperMesh

Connector Authoring

hypermesh.connectors provides a small framework for pulling data from external systems (MDE, S3, webhooks, …) into a hyperedge table on a schedule or on demand.

Terminal window
pip install "hypermesh[connectors]"   # extra deps (e.g. boto3 for S3)

Using a connector

Section titled “Using a connector”
import hypermesh as hm
from hypermesh.connectors import build_connector, ConnectorConfig


cfg = ConnectorConfig(
    id="mde-prod",
    type="mde",
    target_table="MDE_ALERTS",
    credentials={"tenant_id": "…", "client_id": "…", "client_secret": "…"},
)


db = hm.connect("/var/lib/hypermesh/data")
conn = build_connector(cfg, db)


print(conn.test_connection())     # validate creds/reachability
result = conn.sync(limit=10_000)  # pull + map + load
print(result)

Writing your own connector

Section titled “Writing your own connector”

Subclass ConnectorBase and implement the source-specific fetch + mapping. The base class handles batching, the load into the target table, and SyncResult bookkeeping.

from hypermesh.connectors import ConnectorBase, ConnectorConfig


class MySourceConnector(ConnectorBase):
    type = "mysource"


    def test_connection(self) -> bool:
        ...  # ping the source; return True/False


    def fetch(self, *, limit: int | None = None):
        # Yield raw records from your source.
        ...


    def to_hyperedges(self, records):
        # Map raw records → hyperedge dicts:
        #   {"event_ts", "members", "weight", "mean_dist_m", "formation"}
        ...

Register your config type so build_connector can find it, then use it exactly like the built-ins. Keep mapping deterministic so repeated syncs are idempotent.