Multi-hop traversal, ingestion, and real constraints — with a Python API, a JSON one, and Cypher, so an agent and a developer can both use it without being taught anything new.

✨ Highlights

• 🐘 Plain PostgreSQL — two ordinary tables and a recursive CTE. No extension, no sidecar service, no new operational dependency.
• 🧭 Real multi-hop traversal — bounded and unbounded hops, per-hop direction, OPTIONAL, rich JSONB filtering, one round trip.
• 🤖 Three front ends, one engine — Python, JSON (with a ready-made LLM tool schema), and a Cypher subset all compile through the same query builder.
• 🔐 Constraints Neo4j puts behind an enterprise licence — unique, composite, partial, existence, type and CHECK constraints on JSONB properties.
• 🧪 Tested like it matters — SQL-level assertions, a live-Postgres suite, an 85% coverage gate and mutation testing in CI.
• 📊 Measured, not claimed — real benchmark numbers in benchmarks/, including where raw SQL still wins.

⚡ Quick start

pip install hopai

from sqlalchemy import create_engine
from hopai import Graph, Start, Hop, OR, AND, NOT, GT, BETWEEN

graph = Graph(create_engine("postgresql+psycopg2://user:pass@host/db"))

result = graph.traverse(
    Start(where={"type": "person"}),
    Hop(where={"active": True}, via={"kind": "friend"}, hops=(1, 4)),
    Hop(where={"type": "company"}, hops=3),
)

result.nodes            # [{"id": ..., "properties": {...}}, ...]
result.edges            # [{"start_id": ..., "end_id": ..., "properties": {...}}, ...]
result.to_networkx()    # in-memory graph, if you have networkx installed

💡 Why

Most "I need graph queries" projects reach for a dedicated graph database before checking whether they need to. This library is the other answer: if your data already lives in Postgres, a well-indexed recursive CTE handles bounded and unbounded traversal, compound multi-hop patterns, and rich filtering — often faster than a bolted-on graph extension, and competitively with a real graph database, without adding an operational dependency. See benchmarks/ for real, measured numbers, not a claim.

🗄️ Schema

graph.create_schema()   # idempotent; safe to call on every start-up

Two tables — a typed identity column plus a JSONB properties bag on each, and the indexes traversal depends on:

CREATE TABLE nodes (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    properties JSONB NOT NULL DEFAULT '{}'
);
CREATE TABLE edges (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    start_id BIGINT NOT NULL REFERENCES nodes(id),
    end_id   BIGINT NOT NULL REFERENCES nodes(id),
    properties JSONB NOT NULL DEFAULT '{}'
);
CREATE INDEX ON edges (start_id);
CREATE INDEX ON edges (end_id);
CREATE INDEX ON nodes USING GIN (properties);
CREATE INDEX ON edges USING GIN (properties);

BY DEFAULT, not ALWAYS: ids may be supplied or generated.

Different table or column names? Graph(engine, node_table=..., edge_table=..., node_id_col=..., ...).

📥 Getting data in

graph.add_nodes([
    {"id": 1, "type": "person", "name": "Alice"},
    {"type": "company", "name": "Acme"},          # id generated
])
graph.add_edges([
    {"start_id": 1, "end": {"name": "Acme"}, "kind": "works_at", "since": 2019},
])

A row is written one of two ways, and the rule is one line: a row with a properties key is nested; any other row is flat, and every key that isn't an identity key is a property.

{"id": 1, "type": "person"}                    # flat — what you write by hand
{"id": 1, "properties": {"type": "person"}}    # nested — what a traversal returns

The nested form is exactly result.nodes, so a subgraph loads into another graph without reshaping.

Edges take endpoints as start_id/end_id, or as start/end property dicts matching one existing node each — because whatever just wrote the nodes usually doesn't know their generated ids. References are resolved in one batched lookup; matching nothing or several raises.

graph.merge_nodes([{"email": "a@x.com", "name": "Alice"}], on=["email"])

INSERT ... ON CONFLICT DO UPDATE, needing a Unique on the on keys. A match merges the new properties over the old ones and leaves the rest alone (Cypher's ON MATCH SET); replace=True overwrites the bag. Merging is idempotent, which is what makes it the right call for an agent that might retry.

For agents and HTTP handlers, one document, and one schema to hand a model:

from hopai import INGEST_TOOL_SCHEMA

graph.ingest({
    "nodes": [{"id": 1, "type": "person"}],
    "edges": [{"start_id": 1, "end_id": 2, "kind": "knows"}],
})

Nodes are written before edges, so a single document can create a node and an edge that references it. graph.add_networkx(g) loads a networkx graph — the inverse of result.to_networkx().

🔐 Constraints

Neo4j puts uniqueness, composite and existence constraints behind an enterprise licence. Postgres has always had them, and a JSONB property is as constrainable as a column once the expression is indexed:

from hopai import Unique, Required, Check, Index, PropertyType, Col, GT

graph.define_constraints(
    nodes=[
        Required("type"),                            # the key must be present
        Unique("email"),                             # no two nodes share one
        Unique("tenant", "slug"),                    # composite
        Unique("email", where={"type": "person"}),   # only among people
        PropertyType("age", "number"),               # not the string "42"
        Check(GT("age", 0), name="age_positive"),    # any filter, as a CHECK
        Index("type"),                               # plain lookup index
    ],
    edges=[
        Unique(Col("start_id"), Col("end_id"), "kind"),   # one edge of a kind per pair
    ],
)

Idempotent, so it belongs next to create_schema(). A violation raises ConstraintViolation naming the constraint and the offending row rather than a driver error. graph.constraint_ddl(...) returns the exact SQL without running it; graph.drop_constraints(...) is the inverse.

PropertyType is worth the line when a model writes your data: an LLM emitting "42" where you expected 42 breaks every numeric comparison downstream, silently and much later.

where= is the one with no Neo4j equivalent at any price — "email is unique among people" is a partial index, and a partial index is just an index.

Two SQL semantics to know, both of which are what you want once stated:

• A unique index doesn't constrain rows where the property is missing (->>'email' is NULL, and NULLs repeat). Unique("email") means "no two share an email", not "everyone has one" — pair it with Required("email") for both. Neo4j's uniqueness constraint behaves the same way.
• Postgres evaluates CHECK before resolving ON CONFLICT, so a merge row must satisfy every check on its own even when it is destined to update a row that already does.

🔎 Filters

{"type": "person"}                          # equality
{"type": "person", "active": True}          # AND of keys, same dict
{"type": ["person", "company"]}             # OR of values, one key (IN-like)
OR({"type": "person"}, {"type": "company"})
AND(OR(...), {"active": True})
NOT({"type": "person"})                     # includes rows missing the key entirely
GT("age", 18) / GTE / LT / LTE
BETWEEN("age", 18, 65)
lambda col: col.op("~")("^A")               # escape hatch: any real SQLAlchemy expression

A bare list at the top level ([{"a": 1}, {"b": 2}]) raises TypeError rather than being guessed at — it reads ambiguously as "both of these" to a human, when it would have meant OR. Use OR(...) explicitly.

NOT is built on JSONB containment specifically because it handles a missing property correctly (excluded from the positive filter → included under NOT), unlike naive equality-based negation, which treats a missing property as SQL NULL and silently drops it under NOT too. Verified during development to be a real trap, not a hypothetical one — see tests/test_hopai.py::test_not_includes_missing_key.

🧭 Direction and hop count

Hop(hops=3)                 # exactly 3 hops
Hop(hops=(1, 6))            # 1 to 6 hops
Hop(direction="backward")   # follow end_id -> start_id ("what points to this")

Direction is per-hop — a chain can mix forward and backward steps (a "who else does X's dependents depend on" query, for instance).

🧩 OPTIONAL

Hop(where=..., optional=True)

Cypher's OPTIONAL MATCH, equivalent: nodes that reach this point in the chain are kept even if this hop finds nothing for them. Only valid on the last hop — supporting it mid-chain would mean every downstream hop tolerating a missing anchor, a materially larger feature this library hasn't built.

🤖 The JSON interface

For callers that shouldn't or can't write Python — an LLM tool call, an HTTP handler, config-driven traversal:

from hopai import traverse_json

traverse_json(graph, {
    "start": {"where": {"type": "person"}},
    "hops": [
        {"where": {"active": True}, "via": {"kind": "friend"}, "hops": [1, 4]},
        {"where": {"type": "company"}, "hops": 3, "optional": True},
    ],
})

Filters accept the same grammar, spelled as JSON operators: {"and": [...]}, {"or": [...]}, {"not": ...}, {"gt": [key, value]}, {"gte": [...]}, {"lt": [...]}, {"lte": [...]}, {"between": [key, lo, hi]}.

hopai.TRAVERSE_TOOL_SCHEMA is a ready-to-use JSON Schema for wiring this into an LLM function-calling definition directly.

🗣️ Cypher as input syntax

For callers who already think in Cypher — reading and writing:

graph.cypher("""
    CREATE (a:person {email: 'a@x.com'})-[:friend]->(b:person {email: 'b@x.com'})
""")

graph.cypher("""
    MERGE (a:person {email: 'a@x.com'})
    ON CREATE SET a.name = 'Alice'
    ON MATCH SET  a.last_seen = 2026
""")

graph.cypher("""
    MATCH (a:person)-[:friend*1..4]->(b {active: true})
    WHERE b.age > 18
    RETURN b
""")

graph.cypher() returns a Subgraph for a query that reads and an IngestResult for one that writes; traverse_cypher and write_cypher are the same thing when you'd rather be explicit. cypher_to_traversal and graph.cypher_operations show the translation — a (Start, [Hop]) pair, or the ingestion plan — without running anything.

Writes compile to the same add_nodes / merge_nodes / add_edges the Python API calls, in one transaction, with ids from the insert wiring the edges. Three places writes stop short of Cypher:

• MERGE on a whole path is refused. Cypher's MERGE (a {…})-[:x]->(b {…}) matches the entire pattern and creates all of it when it doesn't match, duplicating nodes that already exist. Bind the endpoints first, then MERGE (a)-[:x]->(b).
• MERGE needs a unique index over every property in the pattern — those are the keys Cypher matches on. Anything that shouldn't take part in matching goes in ON CREATE SET. (Cypher needs no index and races instead; the error here names the Unique(...) to declare.)
• MATCH before a write binds single nodes by property, one lookup each. It doesn't traverse.

SET on matched rows, DELETE and DETACH DELETE are unsupported: there's no update-by-query or delete API here yet, in Cypher or in Python.

hopai has no label concept, so labels compile to property tests: (a:person) → {"type": "person"}, [:friend] → {"kind": "friend"}. Change the keys with node_label_key= / edge_type_key=, or pass None to ignore labels entirely.

Translates: linear MATCH chains (including several MATCH clauses joined end to end), *min..max, -> / <- per hop, [:A|B], inline property maps, WHERE with AND/OR/comparisons/IN/IS NULL, all(r IN relationships(p) WHERE ...) → via, and OPTIONAL MATCH as the last clause.

Everything else raises CypherError naming the rewrite, rather than translating into something that answers a different question:

• RETURN has no target. A traversal returns the whole matching subgraph, so projections are parsed and ignored — and aggregations (RETURN count(a)) raise, since the caller clearly wanted a number.
• x.k <> v and NOT x.k = v raise. Cypher evaluates these to NULL when k is missing and drops the row; hopai's containment-based NOT keeps it. Same spelling, different result set. Write the NULL-safe idiom x.k IS NULL OR x.k <> v, which maps exactly onto NOT({"k": v}).
• Also refused: cross-variable OR (a.x = 1 OR b.y = 2), unbounded * (pass max_var_length=N to cap it), undirected -[]-, comma-separated patterns, WITH / ORDER BY / LIMIT, and OPTIONAL MATCH anywhere but last.

🚧 What this doesn't do (yet)

• No disjoint multi-pattern matching (MATCH (a)-[]->(b), (c)-[]->(d) joined on shared variables) — one linear chain of hops only.
• OPTIONAL only on the last hop, not mid-chain.
• Synchronous only — every call blocks; no AsyncSession support yet.
• A cycle-protection path array is carried on every recursive row. Cheap at moderate depth, measurably not-cheap on single-segment traversals past roughly 10 hops — see benchmarks/ for the actual numbers rather than a guess.

🛠️ Development

pip install -e ".[dev]"
docker compose up -d      # throwaway PostgreSQL matching the default DSN
pytest tests/ -v
ruff check .

Most of the suite needs no database at all — query shape, filter compilation and the Cypher translator are all tested against compiled SQL. Those that do need one skip cleanly when it isn't there; set HOPAI_REQUIRE_DB=1 (as CI does) to make a missing database an error instead.

CI enforces a line coverage floor of 85% and runs mutation testing (mutmut) on every PR — a surviving mutant is triaged, not ignored, because a line a mutation can change in silence is a line no test is really asserting on.

📊 Benchmarking

See benchmarks/README.md.

📄 License

MIT — see LICENSE.