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).

🌟 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="genome_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="genome_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.

📚 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
)

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 demo_cypher.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")

📄 License

This project is licensed under the Apache License 2.0.

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