post-graph 0.1.0

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

post-graph: Generic PostgreSQL Graph Database Library

A generic, high-performance mechanism to use PostgreSQL as a graph database in Python. It supports multi-tenancy (realms), automatic shadow audit tables, and advanced graph traversals (recursive CTEs).

Features:

• Table-Per-Vertex & Table-Per-Edge Architecture: Graph elements are mapped to individual tables, utilizing PostgreSQL's native schema constraints and relations.
• Multi-Tenancy: Built-in logical isolation using a realm field, allowing multiple tenants to share the same database while enforcing partition boundaries.
• Shadow Auditing: Automatically creates a matching {table_name}_audit table for every vertex and edge table. Operates at the database trigger level to log operations (INSERT, UPDATE, DELETE), capturing the complete row history and the initiating user_id passed securely via session parameters.
• Dynamic Metadata Discovery: Queries the PostgreSQL catalog tables to dynamically determine the structure and connections of edge tables during traversals.
• Advanced Graph CTEs: High-performance recursive CTE queries for single-hop neighbor queries, multi-hop path traversals, and shortest-path calculation (with early-termination and cycle prevention).
• Multiple Backend Drivers: Includes native clients for raw asyncpg and SQLAlchemy (v2.0 async connections).

---

Architecture Schema

The schema design isolates graphs by tenant realm, ensuring that edges can only link vertices belonging to the exact same realm.

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"

---

Installation

Define dependencies in your environment:

pip install -r requirements.txt

---

Quick Start

1. Raw asyncpg Client (AsyncPostGraph)

import asyncio
from post_graph import AsyncPostGraph

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

    # Create schema (with optional edge-to-vertex cascade deletion parameters)
    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",
        cascade_delete_from=False, # If True, deleting an edge deletes the source vertex
        cascade_delete_to=False    # If True, deleting an edge deletes the target vertex
    )

    # Add vertices (isolated in realm 'tenant_1' and audited as user 'admin_alice')
    await client.add_vertex(
        table_name="users", 
        realm="tenant_1", 
        vertex_id="u_1", 
        payload={"username": "alice"}, 
        user_id="admin_alice"
    )
    await client.add_vertex(
        table_name="posts", 
        realm="tenant_1", 
        vertex_id="p_1", 
        payload={"title": "Hello World"}, 
        user_id="admin_alice"
    )

    # Link vertices with an edge (with optional cycle prevention checks)
    await client.add_edge(
        table_name="likes",
        realm="tenant_1",
        edge_id="e_1",
        from_id="u_1",
        to_id="p_1",
        relation_type="like",
        payload={"reaction": "heart"},
        user_id="user_bob",
        check_cycle=True # Optional: raise CyclicReferenceError if a loop would be created
    )

    # Query neighbors
    neighbors = await client.get_neighbors(
        realm="tenant_1", 
        vertex_table="users", 
        vertex_id="u_1", 
        edge_tables=["likes"]
    )
    for vertex, edge in neighbors:
        print(f"Connected to post: {vertex.id} with payload: {vertex.payload}")

    # Delete an entire realm (tenant) from all tables (including audits)
    deleted_rows = await client.delete_realm("tenant_1")
    print(f"Cleared {deleted_rows} rows belonging to tenant_1")

    await client.close()

asyncio.run(main())

2. SQLAlchemy Client (SQLAlchemyPostGraph)

Integrates with your existing SQLAlchemy async architecture:

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

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

# All schema creation and graph operation interfaces match the AsyncPostGraph client
await client.create_vertex_table("users")

Advanced Graph Operations

Direct Object-Level Traversal

If a Vertex is loaded from a client, you can traverse directly from the object using standard OOP calling patterns.

Note: Since from is a reserved keyword in Python, the reverse traversal method is named .from_() (with an underscore).

# Load a vertex
alice_v = await client.get_vertex(table_name="users", realm="tenant_1", vertex_id="u_1")

# 1. Forward traversal
steps = await alice_v.to("follows")
if steps:
    neighbor_v = steps[0].vertex()
    print(f"Alice follows: {neighbor_v.id}")

# 2. Reverse traversal (incoming links)
rev_steps = await neighbor_v.from_("follows")
if rev_steps:
    creator_v = rev_steps[0].vertex()  # returns Alice!
    print(f"Alice was found by reverse traversal: {creator_v.id}")

# 3. Direct Edge Mutations via Objects/Steps
if rev_steps:
    new_edge = await rev_steps[0].add_edge_to(to_id="p_5", edge_table="likes")
    print(f"Added edge: {new_edge.from_id} -> {new_edge.to_id}")

# 4. Direct Vertex/Edge Deletion
# Deleting a vertex automatically cascade-deletes all connecting edges in the database
await neighbor_v.delete(user_id="admin_1")

Path Traversal

Retrieve all reachable paths up to a specified depth (default 3) starting from a vertex:

paths = await client.traverse(
    realm="tenant_1", 
    start_table="users", 
    start_id="u_1", 
    edge_tables=["likes", "follows"], 
    max_depth=3
)
for p in paths:
    print(f"Path taken: {p['path']} via relations: {p['edge_path']}")

Shortest Path

Find the shortest path between two vertices, preventing cycles:

sp = await client.shortest_path(
    realm="tenant_1",
    start_table="users",
    start_id="u_1",
    target_table="posts",
    target_id="p_5",
    edge_tables=["follows", "likes"]
)
if sp:
    print(f"Shortest path depth: {sp['depth']}, steps: {sp['path']}")

---

Shadow Auditing & Session User Mapping

Auditing works entirely inside PostgreSQL triggers. To attribute a change to a user:

1. When calling Python operations (add_vertex, upsert_edge, delete_vertex, etc.), pass the user_id keyword argument.
2. The library executes SELECT set_config('app.current_user_id', user_id, true) in the current transaction block.
3. The database trigger automatically maps this value into the {table_name}_audit table. A query to {table_name}_audit returns the modification history:

SELECT audit_id, realm, action, changed_by, changed_at, old_row, new_row 
FROM users_audit;

---

Schema-Per-Realm (Physical Namespace Isolation)

By default, post-graph uses a single shared set of tables and isolates data logically using a realm column. If you require physical data isolation, enable the schema_per_realm parameter on client initialization:

# Initialize Async Client with physical namespace isolation
client = AsyncPostGraph(dsn=db_url, schema_per_realm=True)

# Create vertex table under specific realm schema namespace 'tenant_a'
await client.create_vertex_table("users", realm="tenant_a")

When enabled:

1. Every realm has its own dedicated schema namespace (e.g. CREATE SCHEMA IF NOT EXISTS "tenant_a").
2. All vertex, edge, and shadow audit tables for a realm are stored physically inside that schema (e.g., "tenant_a"."users" and "tenant_a"."users_audit").
3. Database level triggers are completely isolated per schema namespace.
4. Schema-specific metadata catalog lookups verify table links locally inside the target schema namespace.