datus-storage-postgresql

PostgreSQL storage adapter for datus-agent. Provides both RDB and Vector backends powered by a single PostgreSQL instance.

Backends

RDB Backend — PostgresRdbBackend

Implements BaseRdbBackend using psycopg v3 and psycopg-pool with a three-layer architecture:

• PostgresRdbBackend (lifecycle): initialize(), connect(namespace, store_db_name), close()
• PgRdbDatabase (database-level, implements RdbDatabase): ensure_table(), transaction(), close()
• PgRdbTable (table-level, implements RdbTable): insert(), query(), update(), delete(), upsert()

Features:

• Full CRUD via PgRdbTable (no need to pass table name)
• upsert() with PostgreSQL ON CONFLICT (dataclass record input)
• transaction() context manager with auto-commit/rollback
• Namespace-based data isolation via PostgreSQL schemas
• Connection pooling with configurable min/max size
• Convenience methods on PgRdbDatabase: get_connection(), execute(), execute_query(), execute_insert()

Vector Backend — PgvectorBackend

Implements BaseVectorBackend using the pgvector extension with a three-layer architecture:

• PgvectorBackend (lifecycle): initialize(), connect(namespace), build_embedding_config(), close()
• PgVectorDb (database-level, implements VectorDatabase): table_exists(), table_names(), create_table(), open_table(), drop_table()
• PgVectorTable (table-level, implements VectorTable): add(), merge_insert(), delete(), update(), search_vector(), search_hybrid(), search_fts(), search_all(), count_rows(), index operations

Features:

• WhereExpr support (condition AST nodes or None) via build_where()
• Vector similarity search (cosine / L2 / inner product)
• Automatic embedding computation on insert
• HNSW vector index, B-tree scalar index, native GIN full-text indexes
• Weighted multi-field FTS with simple, whitespace, raw, and Unicode ngram field configurations
• Strict FTS index version/status checks and automatic transactional index maintenance on incremental writes
• PyArrow Schema to PostgreSQL DDL mapping

Configuration

Both backends register as type: postgresql and accept the same configuration parameters. They can point to the same PostgreSQL instance.

storage:
  rdb:
    type: postgresql
    host: localhost
    port: 5432
    user: postgres
    password: postgres
    dbname: datus
    pool_min_size: 1
    pool_max_size: 10
  vector:
    type: postgresql
    host: localhost
    port: 5432
    user: postgres
    password: postgres
    dbname: datus
    pool_min_size: 1
    pool_max_size: 10

Parameters

Extensions are enabled lazily. Vector tables require vector, while raw substring FTS fields require pg_trgm. Native simple, whitespace, and ngram FTS do not require an extension, and text-only FTS does not require pgvector.

Usage

RDB Backend

from dataclasses import dataclass
from datus.storage.rdb.base import TableDefinition, ColumnDef

@dataclass
class User:
    id: int = None
    name: str = None
    email: str = None

backend = PostgresRdbBackend()
backend.initialize(config)

# connect() returns a RdbDatabase handle (namespace maps to PG schema)
db = backend.connect(namespace="my_app", store_db_name="user_store")

# ensure_table() returns a RdbTable handle
users_table = db.ensure_table(TableDefinition(
    table_name="users",
    columns=[
        ColumnDef(name="id", col_type="INTEGER", primary_key=True, autoincrement=True),
        ColumnDef(name="name", col_type="TEXT"),
        ColumnDef(name="email", col_type="TEXT"),
    ],
))

# Table-level CRUD (no need to pass table name)
row_id = users_table.insert(User(name="Alice", email="alice@example.com"))
users = users_table.query(User, where={"name": "Alice"})
users_table.update({"email": "new@example.com"}, where={"name": "Alice"})
users_table.delete(where={"name": "Alice"})

# Transaction on database level
with db.transaction():
    users_table.insert(User(name="Bob", email="bob@example.com"))
    users_table.insert(User(name="Carol", email="carol@example.com"))

Vector Backend

from datus.storage.conditions import eq, and_

backend = PgvectorBackend()
backend.initialize(config)

# connect() returns a VectorDatabase handle
db = backend.connect(namespace="my_namespace")

# create_table() / open_table() return VectorTable handles
table = db.create_table("my_table", schema=my_schema, embedding_function=emb_config)
table = db.open_table("my_table")

# Table-level operations (no handle passing)
table.add(df)
results = table.search_all(where=eq("category", "active"))
results = table.search_all(where=and_(eq("status", "active"), eq("type", "A")))
results = table.search_vector(query_text="hello", vector_column="vector", top_n=10)

# Database-level operations
db.drop_table("my_table", ignore_missing=True)
assert db.table_exists("my_table") == False

Entry Points

[project.entry-points."datus.storage.rdb"]
postgresql = "datus_storage_postgresql.rdb:register"

[project.entry-points."datus.storage.vector"]
postgresql = "datus_storage_postgresql.vector:register"

Once installed, datus-agent discovers and registers both backends automatically — no manual wiring needed.

Source Layout

datus_storage_postgresql/
├── rdb/
│   ├── __init__.py          # register() → RdbRegistry
│   └── backend.py           # PostgresRdbBackend
└── vector/
    ├── __init__.py           # register() → VectorRegistry
    ├── backend.py            # PgvectorBackend
    └── schema_converter.py   # PyArrow Schema → PostgreSQL DDL