Python API Reference

Install the package once with pip install -e ., then:

import hypermeshdb

---

hypermeshdb.connect()

def connect(
    dir_path:       str,
    hyperedges_csv: str | None = None,
    nodes_csv:      str | None = None,
    bucket_seconds: int = 10,
) -> Connection

Open (or create) a HyperMesh DB database at dir_path.

Parameters

Name	Type	Description
dir_path	str	Filesystem path to the database directory. Created if absent.
hyperedges_csv	str | None	Path to a hyperedges CSV. If the TPI+FMI index does not exist, it is built from this file. Required on first open.
nodes_csv	str | None	Path to a node-properties CSV. Imported to nodes.db on first open; subsequent opens reuse the existing SQLite file.
bucket_seconds	int	TPI bucket granularity in seconds. Ignored if the index already exists. Default: 10.

Returns Connection

Raises HyperMeshError — if the index does not exist and hyperedges_csv is not provided, or on any I/O failure.

Example

# First open — build index
db = hypermeshdb.connect(
    "/tmp/demo",
    hyperedges_csv="data/hyperedges.csv",
    nodes_csv="data/nodes.csv",
)

# Subsequent opens — reuse existing index
with hypermeshdb.connect("/tmp/demo") as db:
    ...

---

Connection

class Connection:
    dir_path:       str
    total_records:  int     # property
    bucket_count:   int     # property
    bucket_seconds: int     # property
    node_count:     int     # property
    wal_pending:    int     # property

An open connection to a database directory. Use as a context manager or call close() explicitly.

execute()

def execute(
    query:      str,
    parameters: dict[str, Any] | None = None,
) -> QueryResult

Execute a Cypher-like query. See the Cypher Reference for supported statement types.

Parameters

Name	Type	Description
query	str	Cypher query string
parameters	dict | None	{"name": value} map. Values are substituted for $name in the query.

Returns QueryResult

Raises HyperMeshError on parse errors or I/O failures.

Example

result = db.execute(
    "MATCH HYPEREDGE (he:CoProximity) "
    "WHERE he.event_ts >= $start AND he.event_ts <= $end RETURN *",
    parameters={"start": 0, "end": 100},
)
for row in result:
    print(row["event_ts"], row.members)

insert()

def insert(
    event_ts:    int,
    members:     list[int],
    weight:      float = 0.0,
    mean_dist_m: float = 0.0,
    formation:   str   = "",
) -> None

Write a new hyperedge to the WAL. Visible to queries immediately.

Raises HyperMeshError if members is empty or exceeds 64 entries.

delete()

def delete(event_ts: int, members: list[int]) -> None

Write a DELETE tombstone to the WAL. The matching record is hidden from all subsequent queries. Physical removal happens at the next compact() call.

Member order is ignored for matching purposes.

compact()

def compact() -> None

Rebuild TPI + FMI from (index ∪ WAL inserts − WAL tombstones), write new binary files atomically, and truncate the WAL.

After compact() returns:

• wal_pending == 0
• All inserted records are permanently in the TPI
• All deleted records are permanently absent

close()

def close() -> None

Release all C and SQLite resources. Safe to call multiple times.

---

QueryResult

class QueryResult:
    columns:    list[str]       # property
    num_tuples: int             # property
    query_plan: QueryPlan | None # property

Iterable result set returned by execute().

Iteration

for row in result:
    print(row["event_ts"])

fetchall()

def fetchall() -> list[Row]

Return all remaining rows and advance the cursor to the end.

fetchone()

def fetchone() -> Row | None

Return the next row and advance the cursor. Returns None when exhausted.

reset()

def reset() -> None

Reset the cursor to the beginning.

to_dicts()

def to_dicts() -> list[dict[str, Any]]

Return all rows as plain Python dicts.

to_df()

def to_df() -> pandas.DataFrame

Convert to a pandas DataFrame. Requires pandas to be installed.

---

Row

A single result row. Supports column-name, integer-index, and attribute access.

row["event_ts"]   # by column name
row[0]            # by integer index
row.event_ts      # by attribute
row.to_dict()     # → dict
row.keys()        # → list[str]
row.values()      # → list[Any]
row.items()       # → list[tuple[str, Any]]

---

QueryPlan

@dataclass(frozen=True)
class QueryPlan:
    strategy:        str    # "TPI_BUCKET_PUSHDOWN" or "FULL_SCAN"
    buckets_scanned: int
    total_buckets:   int
    speedup_factor:  float
    elapsed_us:      int

Performance metadata attached to every data-query result. None for DDL results.

print(result.query_plan)
# TPI_BUCKET_PUSHDOWN: read 5/86 buckets (17.2x speedup, 412µs)

---

HyperMeshError

class HyperMeshError(Exception): ...

Raised for all HyperMesh DB errors: parse failures, I/O errors, schema violations, and invalid arguments.

from hypermeshdb import HyperMeshError

try:
    db.execute("INVALID SYNTAX")
except HyperMeshError as exc:
    print(f"Query failed: {exc}")