hyperweave-db 0.1.1

No project description provided

Hyperweave

A lightweight hypergraph database built on SQLite + SQLAlchemy — weave your data together.

Installation

pip install hyperweave

Or install from source:

git clone <repo-url>
cd hyperweave
pip install -e .

Quick Start

from hyperweave import HypergraphDB

# Open or create a database
db = HypergraphDB("demo.db")

# Add nodes
db.add_node("alice", label="Alice")
db.add_node("bob",   label="Bob")
db.add_node("carol", label="Carol")

# Create a hyperedge (connecting 2+ nodes)
db.create_hyperedge("team_a", ["alice", "bob", "carol"], label="Team A")

# Query
print(db.get_members("team_a"))          # ['alice', 'bob', 'carol']
print(db.get_hyperedges_of("alice"))     # ['team_a']
print(db.get_node_degree("alice"))       # 1

# Visualize (opens in browser)
db.draw()

Database Path Resolution

How HypergraphDB(db_path) resolves the path:

Argument	Storage Location
"my.db" (bare filename)	~/.hyperweave/data/my.db
"./data/my.db" (contains directory)	Relative to current working directory
"/abs/path/to/my.db" (absolute path)	As-is
Env var HYPERWEAVE_PATH	Overrides the default ~/.hyperweave/data

API Reference

Node Operations

add_node(node_id, label="", attrs=None)

Add or update a node (updates label/attrs if the id already exists).

db.add_node("alice")
db.add_node("bob", label="Bob Chen")
db.add_node("carol", label="Carol", attrs={"email": "[e-mail]"})

get_node(node_id) -> dict | None

Get a node's info, or None if not found.

node = db.get_node("alice")
# Returns: {"id": "alice", "label": "Alice", "attrs": {}}

delete_node(node_id)

Delete a node and remove it from all hyperedges.

db.delete_node("alice")

Hyperedge Operations

create_hyperedge(edge_id, member_ids, label="", attrs=None, if_exists="error")

Create a hyperedge connecting the given node IDs. Auto-creates any member nodes that don't exist yet.

• if_exists="error" — raise ValueError (default)
• if_exists="skip" — silently do nothing
• if_exists="replace" — delete the old edge and recreate with new members

db.create_hyperedge("paper_nlp", ["alice", "bob", "carol"],
                    label="NLP Paper",
                    attrs={"year": 2025})

get_hyperedge(edge_id) -> dict | None

Get a hyperedge's info including its member list, or None.

e = db.get_hyperedge("paper_nlp")
# Returns: {
#   "id": "paper_nlp",
#   "label": "NLP Paper",
#   "members": ["alice", "bob", "carol"],
#   "attrs": {"year": 2025}
# }

add_members(edge_id, member_ids)

Append new members to an existing hyperedge.

db.add_members("paper_nlp", ["diana"])

remove_members(edge_id, member_ids)

Remove specific members from a hyperedge.

db.remove_members("paper_nlp", ["bob"])

delete_hyperedge(edge_id)

Delete a hyperedge and all its membership records.

db.delete_hyperedge("paper_nlp")

Queries (LRU cached)

The following queries are @lru_cache-backed, optimized for read-heavy workloads. Any write operation automatically clears the cache.

get_members(edge_id) -> list[str]

Get all node IDs in a hyperedge (ordered by position).

members = db.get_members("paper_nlp")  # ['alice', 'bob', 'carol']

get_hyperedges_of(node_id) -> list[str]

Get all hyperedge IDs that a node belongs to.

edges = db.get_hyperedges_of("alice")  # ['paper_nlp', 'team_a']

get_node_degree(node_id) -> int

Number of hyperedges a node belongs to.

deg = db.get_node_degree("alice")  # 2

get_hyperedge_arity(edge_id) -> int

How many members a hyperedge contains.

arity = db.get_hyperedge_arity("paper_nlp")  # 3

Bulk Export

all_nodes() -> list[dict]

Return every node.

nodes = db.all_nodes()
# [{"id": "alice", "label": "Alice", "attrs": {}}, ...]

all_hyperedges() -> list[dict]

Return every hyperedge with its member list.

edges = db.all_hyperedges()
# [{"id": "paper_nlp", "label": "NLP Paper", "members": [...], "attrs": {}}, ...]

Visualization

draw(port=None, open_browser=True) -> str

Generate a D3.js force-directed hypergraph visualization (convex hulls + nodes + labels).

• draw() — write index.html and open it in the browser
• draw(port=8080) — start an HTTP server (background thread), then open in browser

Returns the file path or URL.

url = db.draw(port=8080)  # Opens http://127.0.0.1:8080/

Cache Control

db.set_modified()  # Manually flush the LRU cache (e.g. after an external write)

Full Example

from hyperweave import HypergraphDB

db = HypergraphDB("example.db")

# Researchers
researchers = ["Alice", "Bob", "Charlie", "Diana", "Eve"]
for r in researchers:
    db.add_node(r)

# Co-authorship as hyperedges
db.create_hyperedge("paper_nlp", ["Alice", "Bob", "Charlie"],
                    label="NLP Paper")
db.create_hyperedge("paper_cv",  ["Alice", "Diana", "Eve"],
                    label="CV Paper")
db.create_hyperedge("paper_db",  ["Bob", "Charlie", "Diana"],
                    label="DB Paper")

# Query: authors of a paper
print(db.get_members("paper_nlp"))     # ['Alice', 'Bob', 'Charlie']

# Query: papers a person co-authored
print(db.get_hyperedges_of("Alice"))   # ['paper_nlp', 'paper_cv']

# Query: who has the most collaborations
for r in researchers:
    deg = db.get_node_degree(r)
    print(f"  {r}: {deg} papers")

# Visualize
db.draw(port=8080)
input("Press Enter to exit...")

Notes

1. Unique hyperedge IDs — create_hyperedge() uses id as the primary key; duplicate calls raise ValueError when if_exists="error" (default). Check with get_hyperedge() first or catch the exception.
2. Auto-create nodes — create_hyperedge() and add_members() auto-create any member nodes that don't exist yet.
3. LRU cache — get_members() and get_hyperedges_of() are LRU-cached. Writes flush the cache automatically; call set_modified() after external writes.
4. Thread safety — check_same_thread=False is set, but single-writer / multi-reader is recommended.
5. Visualization port — draw(port=8080) starts from the given port and auto-increments if occupied (up to 5 attempts).