post-graph 0.1.4

High-performance PostgreSQL-backed graph database library supporting multi-tenant realms, schema-per-realm, shadow auditing, and append-only data history.

post-graph: PostgreSQL-Backed Graph Database Library

A generic, high-performance library for using PostgreSQL as a graph database in Python. It supports multi-tenancy (realms), automatic shadow audit logging, append-only history tables, and high-performance graph traversals (recursive CTEs).

---

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.
• Dual Isolation Modes:
  • Single-Schema Multi-Tenancy: Logical isolation using a realm column partition.
  • Schema-Per-Realm: Physical isolation creating dedicated PostgreSQL schema namespaces per tenant (CREATE SCHEMA IF NOT EXISTS "realm_name").
• 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 the 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, id, payload, timestamp).
  • Cascades deletion when main vertex/edge is deleted (ON DELETE CASCADE).
• Shadow Audit Tables ({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:
  • Recursive CTE queries for neighbor exploration, path discovery, and cycle-free shortest path calculation.
  • Direct object-oriented traversal APIs (vertex.to(), vertex.from_(), step.vertex(), step.add_edge_to()).
• Multiple Async Client Drivers:
  • Raw 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]"

---

Architecture Schema

erDiagram
    users {
        text realm PK
        bigserial id PK
        text fqid
        jsonb payload
        timestamptz created_at
        timestamptz updated_at
    }
    posts {
        text realm PK
        bigserial id PK
        text fqid
        jsonb payload
        timestamptz created_at
        timestamptz updated_at
    }
    likes {
        text realm PK, FK
        bigserial id PK
        text fqid
        bigint from_id FK
        bigint to_id FK
        text relation_type
        jsonb payload
        timestamptz created_at
        timestamptz updated_at
    }
    users_audit {
        bigint audit_id PK
        text realm
        text action
        text changed_by
        timestamptz changed_at
        jsonb old_row
        jsonb new_row
    }
    users_data {
        bigserial data_id PK
        text realm FK
        bigint id FK
        jsonb payload
        timestamptz timestamp
    }

    users ||--o{ likes : "from_id"
    posts ||--o{ likes : "to_id"
    users ||--o{ users_audit : "logs changes"
    users ||--o{ users_data : "appends records"

---

Quick Start Examples

1. Using AsyncPostGraph (asyncpg)

import asyncio
from post_graph import AsyncPostGraph

async def main():
    client = AsyncPostGraph(dsn="postgresql://postgres:postgres@localhost:5432/postgres")
    await client.connect()

    # 1. Create schema tables
    await client.create_vertex_table("users")
    await client.create_vertex_table("posts")
    await client.create_edge_table("likes", from_vertex_table="users", to_vertex_table="posts")

    # 2. Add vertices (id is autogenerated if omitted)
    alice = await client.add_vertex("users", realm="realm_a", payload={"name": "Alice"})
    post = await client.add_vertex("posts", realm="realm_a", payload={"title": "Hello PostGraph!"})

    # 3. Add edge connecting Alice -> Post
    edge = await client.add_edge(
        table_name="likes",
        realm="realm_a",
        from_id=alice.id,
        to_id=post.id,
        relation_type="like",
        payload={"reaction": "heart"}
    )

    # 4. Append data history record
    await alice.add_data({"status": "active", "login_count": 1})
    history = await alice.get_data()
    print(f"Alice history count: {len(history)}, last: {history[0].payload}")

    # 5. Object-oriented traversal
    steps = await alice.to("likes")
    for step in steps:
        target_post = step.vertex()
        print(f"Alice liked post: {target_post.payload['title']}")

    await client.close()

asyncio.run(main())

2. Using SQLAlchemyPostGraph

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from post_graph import SQLAlchemyPostGraph

async def main():
    engine = create_async_engine("postgresql+asyncpg://postgres:postgres@localhost:5432/postgres")
    client = SQLAlchemyPostGraph(engine)

    # DDL & CRUD methods share identical signatures
    await client.create_vertex_table("users")
    await client.create_edge_table("follows", from_vertex_table="users", to_vertex_table="users")

    v1 = await client.add_vertex("users", realm="default", payload={"name": "Bob"})
    v2 = await client.add_vertex("users", realm="default", payload={"name": "Charlie"})
    e = await client.add_edge("follows", realm="default", from_id=v1.id, to_id=v2.id, relation_type="follow")

    await engine.dispose()

asyncio.run(main())

---

Detailed API Reference

Client Initialization & Connection

AsyncPostGraph(
    dsn: Optional[str] = None,
    connection_or_pool: Optional[Union[asyncpg.Pool, asyncpg.Connection]] = None,
    schema_per_realm: bool = False
)

SQLAlchemyPostGraph(
    engine_or_connection: Union[AsyncEngine, AsyncConnection],
    schema_per_realm: bool = False
)

• schema_per_realm: If True, creates physically isolated schema namespaces per realm (e.g. "realm_a"."users"). If False (default), uses single-schema logical multi-tenancy with a realm column partition.

---

Schema DDL Operations

create_vertex_table

await client.create_vertex_table(table_name="users", realm=None)

Creates:

1. {table_name} main vertex table with (realm, id) primary key.
2. {table_name}_audit shadow audit log table.
3. {table_name}_data append-only history table with foreign key (realm, id) ON DELETE CASCADE.
4. Triggers for updated_at timestamps and audit logging.

create_edge_table

await client.create_edge_table(
    table_name="likes",               # Optional; defaults to "{from_table}TO{to_table}" if empty
    from_vertex_table="users",
    to_vertex_table="posts",
    realm=None,
    cascade_delete_from=False,       # If True, deleting the edge deletes source vertex
    cascade_delete_to=False          # If True, deleting the edge deletes target vertex
)

Creates main edge table, shadow audit table {table_name}_audit, and append-only history table {table_name}_data. Foreign keys reference (realm, from_id) and (realm, to_id) on vertex tables with ON DELETE CASCADE.

---

Vertex Operations

add_vertex

vertex = await client.add_vertex(
    table_name="users",
    realm="tenant_1",
    vertex_id=None,                 # Optional. If omitted, BIGSERIAL autogenerates integer ID.
    payload={"name": "Alice"},       # Dict payload stored as JSONB.
    user_id="admin_1"               # Optional user attribution for audit log.
)

get_vertex

vertex = await client.get_vertex(table_name="users", realm="tenant_1", vertex_id=1)

update_vertex

updated = await client.update_vertex(
    table_name="users",
    realm="tenant_1",
    vertex_id=1,
    payload={"name": "Alice Smith"}, # Replaces JSONB payload
    user_id="admin_1"
)

delete_vertex

deleted = await client.delete_vertex(table_name="users", realm="tenant_1", vertex_id=1, user_id="admin_1")

Automatically cascade deletes all connected edges in the database.

---

Edge Operations

add_edge

edge = await client.add_edge(
    table_name="follows",
    realm="tenant_1",
    edge_id=None,                   # Optional autogenerated BIGSERIAL ID.
    from_id=1,
    to_id=2,
    relation_type="friend",
    payload={"since": "2026-01-01"},
    user_id="admin_1",
    check_cycle=False               # Set True to raise CyclicReferenceError if a cycle is created
)

get_edge

edge = await client.get_edge(table_name="follows", realm="tenant_1", edge_id=100)

delete_edge

deleted = await client.delete_edge(table_name="follows", realm="tenant_1", edge_id=100, user_id="admin_1")

---

Multi-Tenant Realm Operations

delete_realm

rows_deleted = await client.delete_realm(realm="tenant_to_remove")

Completely removes all vertices, edges, audit logs, and data history records belonging to the target realm across all tables.

---

Append-Only Data History ({table_name}_data)

Vertices and edges support appending timestamped historical data snapshots.

Client Data APIs

# Add historical record
record = await client.add_vertex_data(
    table_name="users",
    realm="realm_a",
    vertex_id=1,
    payload={"status": "active", "login_count": 10},
    timestamp=None                  # Optional; defaults to CURRENT_TIMESTAMP
)

# Fetch history records (sorted timestamp DESC)
history = await client.get_vertex_data(table_name="users", realm="realm_a", vertex_id=1, limit=5)
for rec in history:
    print(rec.data_id, rec.timestamp, rec.payload)

(Direct add_edge_data and get_edge_data methods operate identically for edge tables).

Model Convenience Methods

# Vertex/Edge instances expose add_data() and get_data() directly
rec = await vertex.add_data({"status": "active"})
history = await vertex.get_data(limit=10)

rec_edge = await edge.add_data({"event": "click"})
history_edge = await edge.get_data()

---

Object-Oriented Graph Traversal

Loaded Vertex objects allow seamless graph navigation:

alice = await client.get_vertex("users", "realm_a", 1)

# 1. Forward Traversal (.to)
steps = await alice.to(edge_table="follows")
for step in steps:
    edge = step.edge
    neighbor = step.vertex()
    print(f"Alice --[{edge.relation_type}]--> {neighbor.payload['name']}")

# 2. Reverse Traversal (.from_)
bob = await client.get_vertex("users", "realm_a", 2)
inbound_steps = await bob.from_(edge_table="follows")
for step in inbound_steps:
    origin_user = step.vertex()
    print(f"Followed by: {origin_user.payload['name']}")

# 3. Add Edge via Traversal Step
new_edge = await inbound_steps[0].add_edge_to(to_id=3, edge_table="follows", relation_type="knows")

# 4. Delete Vertex directly
await bob.delete(user_id="admin_1")

---

High-Performance Recursive CTE Queries

traverse

Explores all reachable paths starting from a vertex up to max_depth:

paths = await client.traverse(
    realm="realm_a",
    start_table="users",
    start_id=1,
    edge_tables=["follows", "likes"],
    max_depth=3
)
for p in paths:
    print(f"Depth: {p['depth']} | Path: {' -> '.join(p['path'])}")

shortest_path

Calculates the shortest path between two nodes, with built-in loop avoidance:

sp = await client.shortest_path(
    realm="realm_a",
    start_table="users",
    start_id=1,
    target_table="posts",
    target_id=42,
    edge_tables=["follows", "likes"],
    max_depth=5
)
if sp:
    print(f"Shortest path found at depth {sp['depth']}: {sp['path']}")

---

Transaction Support

AsyncPostGraph (raw asyncpg)

async with pool.acquire() as conn:
    async with conn.transaction():
        # Bind transaction connection to client wrapper
        tx_client = AsyncPostGraph(connection_or_pool=conn)
        await tx_client.add_vertex("users", "realm_a", payload={"name": "Tx Node"})
        await tx_client.add_edge("follows", "realm_a", from_id=1, to_id=2, relation_type="knows")
        # Commits automatically on block exit; rolls back on exception.

SQLAlchemyPostGraph

async with engine.connect() as conn:
    async with conn.begin():
        tx_client = SQLAlchemyPostGraph(conn)
        await tx_client.add_vertex("sa_users", "default", payload={"name": "Tx Node"})
        await tx_client.add_edge("sa_follows", "default", from_id=1, to_id=2, relation_type="knows")

---

License

This project is licensed under the MIT License.