post-graph: PostgreSQL-Backed Graph Database Library

A high-performance Python library for using PostgreSQL as a native graph database. It supports multi-tenant realms, application-level space sub-grouping (space), pgvector similarity search across main & history tables, automatic shadow audit logging, append-only history tables, and high-speed recursive graph traversals (CTEs).

🧪 Runnable Examples

Ten scripts in examples/, each one capability. They need PostgreSQL and nothing else — no LLM, no API keys:

export POSTGRES_URI=postgresql://localhost:5432/postgres
cd examples && python 01_quickstart.py

Every one was executed against a live database before being committed.

🌟 Key Features

• Table-Per-Vertex & Table-Per-Edge Architecture: Maps graph elements directly to relational tables, taking advantage of PostgreSQL's foreign keys, indexes, and constraints.
• Hierarchical Multi-Tenancy & Space Sub-grouping:
  • Macro-Isolation (realm): Primary tenant partitioning at database or schema level.
  • Micro-Isolation (space): Optional application-level sub-grouping (space VARCHAR(255) DEFAULT 'default') within tables to segregate environments (e.g. production, staging, sandbox, or workspace spaces).
  • Dual Tenant Topologies:
    • Single-Schema Multi-Tenancy: Logical isolation using realm & space column partitions.
    • Schema-Per-Realm: Physical isolation creating dedicated PostgreSQL schema namespaces per tenant (CREATE SCHEMA IF NOT EXISTS "realm_name").
• pgvector Similarity Search (vector_search):
  • Native HNSW vector indexing (vector(dim)) for vertex embeddings.
  • Cosine distance (<=>), Euclidean L2 (<->), and Inner Product (<#>).
  • Multi-scope searching across main tables, associated data history tables, or both combined (search_scope="both").
• Autogenerated BIGSERIAL Primary Keys & Computed fqid:
  • Vertex FQID: {realm}/{table_name}/{id}
  • Edge FQID: {realm}/{from_table}-{to_table}/{id} (using hyphen separator)
  • Automatically populated at PostgreSQL level via GENERATED ALWAYS AS ... STORED NOT NULL.
• Append-Only History Tables ({table_name}_data):
  • Automatically created alongside vertex and edge tables.
  • Stores timestamped JSONB payload updates (data_id, realm, space, id, payload, timestamp, embedding).
  • Supports vector embeddings on data records for historical semantic matching.
  • Cascades deletion when main vertex/edge is deleted (ON DELETE CASCADE).
• Shadow Audit Logging ({table_name}_audit):
  • Operates via PostgreSQL triggers to capture all INSERT, UPDATE, and DELETE events.
  • Logs old/new row state and the initiating user_id passed via session parameters (app.current_user_id).
• Advanced Graph Traversals & Cycle Detection:
  • Recursive CTE queries for neighbor exploration, path discovery, and cycle-free shortest path calculation.
  • Optional check_cycle=True raising CyclicReferenceError during edge creation.
  • Direct object-oriented traversal APIs (vertex.to(), vertex.from_(), step.vertex(), step.add_edge_to()).
• Promoted Payload Columns:
  • Hot JSONB keys become generated, indexed columns maintained by PostgreSQL, so the planner can use an index where it previously scanned.
  • valid_from and valid_to are promoted by default as pt_valid_from / pt_valid_to, holding the date normalised to YYYY-MM-DD; any other key can be promoted via promoted_keys as p_{key}.
  • The write path is unchanged — callers keep writing plain JSON — and a table without the columns falls back to payload->>, returning the same rows.
• openCypher Queries (CypherSession):
  • A documented Cypher subset over the same tables: a label is a vertex table, a relationship type is a relation_type value, a property is a payload key.
  • Reads compile to a single SQL statement; writes run through the client's own methods, in one transaction, so audit tables and triggers behave and a failed query leaves nothing behind.
  • Conformance is measured against the openCypher TCK rather than asserted. See docs/cypher.md.
• Multiple Async Client Drivers:
  • High-speed asyncpg client (AsyncPostGraph).
  • SQLAlchemy v2.0 async client (SQLAlchemyPostGraph).

📦 Installation

Install post-graph from PyPI:

# Basic installation (includes asyncpg)
pip install post-graph

# Installation with SQLAlchemy support
pip install "post-graph[sqlalchemy]"

# Installation with all optional dependencies
pip install "post-graph[all]"

Using uv:

uv add post-graph
# or with SQLAlchemy support:
uv add "post-graph[sqlalchemy]"

PostgreSQL Setup

To enable pgvector similarity search, ensure the pgvector extension is enabled in PostgreSQL:

CREATE EXTENSION IF NOT EXISTS vector;

🏗️ Database Architecture & Multi-Tenancy Hierarchy

erDiagram
    REALM ||--o{ SPACE : contains
    SPACE ||--o{ VERTICES : contains
    SPACE ||--o{ EDGES : contains

    agent_registry {
        text realm PK
        bigserial id PK
        varchar_255 space "Indexed optional sub-grouping (e.g. production, sandbox)"
        text fqid "Generated: realm/table/id"
        uuid uuid "Unique UUID identifier"
        jsonb payload "JSON metadata"
        vector embedding "HNSW Cosine Vector"
        timestamptz created_at
        timestamptz updated_at
    }

    agent_registry_data {
        bigserial data_id PK
        text realm FK
        bigint id FK
        varchar_255 space
        jsonb payload
        vector embedding
        timestamptz timestamp
    }

    spawns {
        text realm PK, FK
        bigserial id PK
        varchar_255 space
        bigint from_id FK
        bigint to_id FK
        text relation_type
        jsonb payload
        timestamptz created_at
    }

    agent_registry ||--o{ agent_registry_data : "appends history"
    agent_registry ||--o{ spawns : "from_id / to_id"

🚀 Quick Start Guide

import asyncio
from post_graph import AsyncPostGraph

async def main():
    # 1. Initialize client
    client = AsyncPostGraph(dsn="postgresql://user:password@localhost:5432/mydb")
    await client.connect()

    realm = "proj_alpha"

    # 2. Declare Vertex & Edge tables
    await client.create_vertex_table("agents", realm=realm, vector_dim=1536)
    await client.create_edge_table("spawns", from_vertex_table="agents", to_vertex_table="agents", realm=realm)

    # 3. Upsert Vertices with Space Sub-grouping
    v_parent = await client.upsert_vertex(
        table_name="agents",
        realm=realm,
        vertex_id=1,
        space="production",
        payload={"name": "The Prime Orchestrator", "caste": "genesis"},
        embedding=[0.01] * 1536
    )

    v_progeny = await client.upsert_vertex(
        table_name="agents",
        realm=realm,
        vertex_id=2,
        space="production",
        payload={"name": "Specialized Worker", "caste": "progeny"},
        embedding=[0.02] * 1536
    )

    # 4. Create Directed Edge
    edge = await client.add_edge(
        table_name="spawns",
        realm=realm,
        from_id=v_parent.id,
        to_id=v_progeny.id,
        space="production",
        relation_type="spawned_progeny",
        payload={"reason": "Task Delegation"}
    )

    # 5. Query Vertices by Realm & Space
    prod_agents = await client.get_vertices("agents", realm=realm, space="production")
    print(f"Production agents: {[a.payload['name'] for a in prod_agents]}")

    # 6. Object-Oriented Neighbor Traversal
    neighbors = await v_parent.outgoing("spawns")
    for step in neighbors:
        print(f"Parent -> {step.vertex().payload['name']} (Edge: {step.edge.relation_type})")

    await client.close()

if __name__ == "__main__":
    asyncio.run(main())

⏱️ Range Queries, Ordering and Bulk Deletion

Polling a growing table must not mean fetching it. where pushes range predicates into SQL over the JSONB payload (parameter-bound, key-validated), order_by/limit shape the result server-side, and count_vertices / delete_vertices complete the poll–work–purge loop without transferring rows.

# An event scheduler's tick: the few due, undone events — not the whole table.
due = await pg.find_vertices("events", realm=world,
    where=[("done_at", "is_null", None), ("due_at", "<=", now_str)],
    order_by="due_at", limit=200)

# A work queue's pending slice.
pending = await pg.find_vertices("decision_queue", realm="agents",
    where=[("done_at", "is_null", None)], limit=500)

# Periodic purge of completed history, returning rows deleted.
purged = await pg.delete_vertices("events", realm=world,
    where=[("done_at", "not_null", None), ("done_at", "<", cutoff_str)])

# Queue depth without row transfer; index the hot predicate once at startup.
depth = await pg.count_vertices("decision_queue", realm="agents",
    where=[("done_at", "is_null", None)])
await pg.create_payload_index("events", realm=world, key="due_at")

Ops: = != < <= > >= is_null not_null in. Values always bind as parameters. int/float compare numerically ((payload->>'k')::numeric); str compares as text, so zero-padded sortable strings (fixed-width timestamps) keep their text ordering. is_null matches absent keys and JSON null alike — a scheduler's "not done yet" in one predicate. order_by follows the numeric cast when a numeric predicate references the same key. delete_vertices refuses an empty where: a full wipe must be the explicit delete_realm. create_payload_index(..., numeric=True) indexes the cast expression; idempotent, deterministically named.

🔌 Connection Pooling for Fleets

Each client manages an asyncpg pool. pool_min_size defaults to 1 (lazy): a fleet of N workers holds N idle connections and grows to pool_max_size (default 10) only under load. connect() retries "too many clients" and "cannot connect now" with exponential backoff and jitter (retries=5 by default), so a deploy herd staggers in instead of crashing.

Sizing rule: workers × pool_max_size must stay below the server's max_connections, with headroom for everything else that talks to the database. Past that point no client-side setting helps — put PgBouncer in between.

📚 Comprehensive API Reference

Client Initialization & Management

AsyncPostGraph(dsn, schema_per_realm=False)

Initializes the high-performance asyncpg graph client.

client = AsyncPostGraph(
    dsn="postgresql://postgres:postgres@localhost:5432/postgres",
    schema_per_realm=False  # Set True for physical PostgreSQL schema isolation
)
await client.connect()

SQLAlchemyPostGraph(dsn_or_engine, schema_per_realm=False)

Initializes the SQLAlchemy v2.0 async graph client.

Schema Definition APIs

create_vertex_table(table_name, realm=None, vector_dim=None, temporal_keys=None, promoted_keys=None)

Creates a vertex table, associated audit table ({table_name}_audit), and append-only data history table ({table_name}_data).

await client.create_vertex_table(
    table_name="agents",
    realm="proj_alpha",
    vector_dim=1536,             # Enables pgvector HNSW index
    promoted_keys=["status"],    # Indexed column for a hot payload key
)

temporal_keys overrides which pair of payload keys is promoted as dates (default ('valid_from', 'valid_to')); promoted_keys promotes further keys verbatim. See Promoted Payload Columns below.

create_edge_table(table_name=None, from_vertex_table=..., to_vertex_table=..., cascade_delete_from=False, cascade_delete_to=False, realm=None, vector_dim=None, temporal_keys=None, promoted_keys=None)

Creates a directed edge table linking two vertex tables.

Pass vector_dim to give edges their own pgvector embedding column and HNSW index, enabling vector_search_edges. Optional — most workloads reach edges by traversing from a vertex rather than by similarity.

await client.create_edge_table(
    "relations",
    from_vertex_table="entities",
    to_vertex_table="entities",
    realm=realm,
    vector_dim=1536  # Optional: enables semantic search over relationships
)

Vertex Operations

add_vertex / upsert_vertex

Upserts a vertex object into a specific {realm} and optional {space}.

vertex = await client.upsert_vertex(
    table_name="agents",
    realm="proj_alpha",
    vertex_id=101,                  # Optional numeric ID or FQID
    space="production",             # Optional space sub-grouping (default: 'default')
    payload={"name": "Polymath Node"},
    embedding=[0.05] * 1536,         # Optional vector embedding
    user_id="user_admin"            # Optional for shadow audit attribution
)

update_vertex(table_name, realm, vertex_id, payload, space=None, replace=False)

Changes an existing vertex with a bare UPDATE, and never an INSERT. Returns the updated vertex, or None when no row matches (realm, vertex_id) — it does not create the row.

spent = await client.update_vertex(
    table_name="invitations",
    realm="trial_platform",
    vertex_id=4211517,
    payload={"redeemed_at": "2026-10-10T09:00:00Z"},  # merged, as upsert_vertex merges
)
if spent is None:
    ...  # no such invitation

Use it where the grant is the containment. upsert_vertex emits INSERT … ON CONFLICT DO UPDATE, and PostgreSQL checks the INSERT privilege on that statement unconditionally — not only when the insert path is taken, because the check is on the statement, not on the branch the data happens to take. So a role granted SELECT, UPDATE gets InsufficientPrivilegeError from upsert_vertex on a row it is entitled to change. A caller that may spend a single-use token but must not be able to mint one is granted UPDATE and withheld INSERT deliberately; update_vertex is the write that fits that grant.

Two differences from upsert_vertex worth knowing:

• space defaults to None, meaning leave the space alone. upsert_vertex defaults it to "default" and always writes it, because it may be creating the row. An update that relocated a row to "default" because the caller did not mention a space would be a surprising way to lose a tenant's data. Pass space="…" to move the row deliberately.

• On an audited table the shadow trigger runs INSERT INTO <table>_audit as the invoking role, so a SELECT, UPDATE role still fails — with permission denied for table <table>_audit, naming a table absent from the grant. Two write-only grants fix it without widening the role's reach into the audited table:

  GRANT INSERT ON <table>_audit TO <role>;
  GRANT USAGE ON SEQUENCE <table>_audit_audit_id_seq TO <role>;
  

  The role can then append an audit record but still cannot read the audit table, nor insert into the audited one. The alternative, create_vertex_table(..., audited=False), trades the audit trail away — for a security-relevant write, usually the wrong trade.

get_vertex(table_name, realm, vertex_id)

Fetches a single vertex by its numeric ID, UUID, or FQID.

get_vertices(table_name, realm, space=None, limit=None)

Fetches all vertices belonging to a {realm}, with optional filtering by {space}.

# Fetch only production space vertices
prod_vertices = await client.get_vertices("agents", realm="proj_alpha", space="production")

# Fetch all vertices in realm regardless of space
all_vertices = await client.get_vertices("agents", realm="proj_alpha")

delete_vertex(table_name, realm, vertex_id, user_id=None)

Deletes a vertex and automatically cascades deletion to referencing edges and history records.

Edge Operations

add_edge

Creates a directed edge between from_id and to_id.

edge = await client.add_edge(
    table_name="spawns",
    realm="proj_alpha",
    from_id=101,
    to_id=102,
    space="production",
    relation_type="spawned_progeny",
    payload={"timestamp": "2026-07-27"},
    check_cycle=True                # Raises CyclicReferenceError if edge creates a cycle
)

get_edges(table_name, realm, space=None, limit=None)

Fetches edges in a realm, optionally filtered by {space}.

pgvector Semantic Search (vector_search)

Performs high-speed cosine, L2, or inner product vector search using HNSW indexes.

results = await client.vector_search(
    table_name="agents",
    realm="proj_alpha",
    query_vector=[0.05] * 1536,
    top_k=5,
    distance_metric="cosine",      # 'cosine', 'l2', or 'inner_product'
    search_scope="both"             # 'main', 'data', or 'both'
)

for vertex, distance in results:
    print(f"Agent: {vertex.payload['name']} | Distance: {distance:.4f}")

Edge Semantic Search (vector_search_edges)

Edges can carry embeddings too, when the edge table was created with a vector_dim. Supply the vector on add_edge / upsert_edge, then search:

await client.add_edge(
    "relations", realm=realm,
    from_id=zeus.id, to_id=hera.id,
    relation_type="married_to",
    payload={"description": "spouse"},
    embedding=[0.05] * 1536          # Optional: stored when the table has a vector column
)

results = await client.vector_search_edges(
    table_name="relations",
    realm=realm,
    query_vector=[0.05] * 1536,
    top_k=5,
    distance_metric="cosine",
    space="production",              # Optional space filter
    relation_type="married_to"       # Optional relation type filter
)

for edge, distance in results:
    print(f"{edge.relation_type} | Distance: {distance:.4f}")

If the edge table has no vector column, a supplied embedding is ignored with a warning and vector_search_edges returns [].

Append-Only Data History ({table_name}_data)

Appends timestamped data records to vertices or edges for full auditability and semantic versioning.

# Append historical data record
record = await client.add_vertex_data(
    table_name="agents",
    realm="proj_alpha",
    vertex_id=101,
    space="production",
    payload={"checkpoint": 18, "status": "active"},
    embedding=[0.02] * 1536
)

# Retrieve history records
history = await client.get_vertex_data("agents", realm="proj_alpha", vertex_id=101, limit=10)

Advanced Graph Traversals & Shortest Path (CTEs)

traverse

Executes recursive CTE traversals across edge tables.

paths = await client.traverse(
    realm="proj_alpha",
    start_table="agents",
    start_id=101,
    edge_tables=["spawns", "collaborates"],
    max_depth=4
)
for p in paths:
    print(f"Depth: {p['depth']} | Path: {' -> '.join(p['path'])}")

shortest_path

Finds the shortest cycle-free path between two vertices.

sp = await client.shortest_path(
    realm="proj_alpha",
    start_table="agents",
    start_id=101,
    target_table="agents",
    target_id=105,
    edge_tables=["spawns", "collaborates"],
    max_depth=5
)
if sp:
    print(f"Shortest path found at depth {sp['depth']}: {sp['path']}")

Promoted Payload Columns

Properties live in payload JSONB, which is flexible but opaque to the planner: payload->>'valid_from' <= '2024-01-01' cannot use an index, so the as-of filter — which runs on every step of every traversal — degraded to a sequential scan.

A promoted column fixes that without touching the write path. PostgreSQL maintains it from payload on insert and update, so you keep writing plain JSON:

await client.create_edge_table(
    "relations",
    from_vertex_table="entities", to_vertex_table="entities",
    realm=realm,
    promoted_keys=["t_expired"],     # p_t_expired, indexed
)

# Unchanged: nothing about writing has to know a column exists.
await client.add_edge("relations", realm, a.id, b.id, "WORKS_AT",
                      payload={"valid_from": "2024-06"})

Measured on 65k rows, the temporal filter moves from a 588-buffer sequential scan to a 27-buffer bitmap index scan — about 31ms to 2.5ms warm — and the margin grows with the table.

Three things worth knowing:

• The names are prefixed. valid_from stays in payload; the column beside it is pt_valid_from. Prefixes keep a promoted column from ever colliding with a real column.
• They are database-side only. Vertex and Edge carry no pt_ fields. The column is derived and read-only, and whether it exists depends on when the table was created, so exposing it on the model would offer a field you can read but never assign, present on some objects and not others.
• Existing tables do not gain them retroactively. The DDL runs at table creation. A realm created before this feature keeps working and returns the same rows via payload->>, just without the index.

Temporal columns hold normalised text rather than DATE because casting text to date is only STABLE and PostgreSQL rejects non-immutable expressions in a generated column. ISO-8601 sorts lexically, so text and date comparison agree — including partial dates, where '2024' normalises to '2024-01-01' rather than sorting as a short string.

openCypher Queries

from post_graph import AsyncPostGraph, CypherSession

session = CypherSession(client, realm="my_realm")
rows = await session.run(
    "MATCH (p:Person)-[:KNOWS]->(f:Person) "
    "WHERE p.name = $name AND f.age > 30 "
    "RETURN f.name AS friend ORDER BY friend",
    {"name": "Alice"},
)

A label is a vertex table, a relationship type is a value in relation_type, and a property is a payload key — read through a promoted column when one exists.

Reads compile to one SQL statement. Writes do not: CREATE, MERGE, SET and DELETE go through the client's own methods so audit tables, triggers and realm rules behave as for any other caller, and the whole query runs in one transaction, so a CREATE that fails part-way leaves nothing behind.

session.explain(query) shows what will happen — the SQL for a read, the sequence of client operations for a write — without running it.

The subset is bounded and every boundary raises rather than being approximated; WITH, UNION, OPTIONAL MATCH, path variables and RETURN * are refused. Conformance is measured against the openCypher TCK rather than asserted. See docs/cypher.md for what is supported and the current numbers, and examples/07_opencypher.py for a runnable tour.

Multi-Tenant Realm & Space Deletion

# Delete an entire realm across all tables
deleted_count = await client.delete_realm(realm="tenant_to_remove")

🤝 Contributing

Bug reports, failing test cases and pull requests are all welcome. CONTRIBUTING.md covers the parts that are specific to this project rather than generic advice: what the test suite needs, what it deliberately does not need, and the invariants worth understanding before changing them.

Issues tagged good first issue are real gaps rather than make-work — each one names the file to look at, what "done" means, and what you will learn from it.

Tests need PostgreSQL with pgvector and nothing else — no LLM, no API keys. The same is true of every example.

📄 License

This project is licensed under the Apache License 2.0.

Developed by Chandan Rajah (chandan.rajah@gmail.com).