Python SDK

The Dreambase Python SDK wraps asyncpg with first-class VECTOR(dims) column support: automatic float32 / NumPy array serialization to the wire format, typed query builders with the .near() method, Pydantic v2 model integration via Vector[dims] generic type, and db.explain_hybrid() for programmatic plan inspection. The SDK is optional — any psycopg2 or asyncpg client connects to Dreambase directly using the PostgreSQL wire protocol.

Installation

pip install dreambase

# With Pydantic model support
pip install "dreambase[pydantic]"

Requirements: Python 3.9+, numpy, asyncpg. All installed automatically.

Client initialization

import dreambase

# Synchronous client
db = dreambase.connect(
    "postgresql://user:[email protected]/myapp"
)

# With explicit options
db = dreambase.connect(
    host="cluster.dreambase.io",
    port=5432,
    user="myuser",
    password="mypass",
    database="myapp",
    ssl=True,
    min_connections=2,
    max_connections=10
)

# Context manager
with dreambase.connect(url) as db:
    rows = db.query("SELECT 1")

Table operations

# Create hybrid table
db.execute("""
    CREATE TABLE IF NOT EXISTS documents (
        id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
        user_id TEXT NOT NULL,
        content TEXT,
        embedding VECTOR(1536),
        created_at TIMESTAMPTZ DEFAULT NOW()
    )
""")

# Insert row with vector
import numpy as np
vec = np.array(my_model.encode(text), dtype=np.float32)
db.execute(
    "INSERT INTO documents (user_id, content, embedding) VALUES ($1, $2, $3)",
    "u_441", "Document text", vec
)

# Batch insert (efficient)
rows = [(uid, content, vec) for uid, content, vec in batch]
db.executemany(
    "INSERT INTO documents (user_id, content, embedding) VALUES ($1, $2, $3)",
    rows
)

Query builder

from dreambase import Q

# Fluent query builder
results = (
    db.from_table("documents")
      .select("id", "content", "created_at")
      .where(Q.eq("user_id", "u_441"))
      .near("embedding", query_vec)
      .limit(5)
      .fetch()
)

# Raw SQL (always available)
results = db.query(
    "SELECT id, content FROM documents WHERE user_id=$1 ORDER BY embedding NEAR $2 LIMIT 5",
    "u_441", query_vec
)

Async support

import asyncio
import dreambase

async def search(user_id: str, query_vec):
    async with dreambase.AsyncClient(url=DATABASE_URL) as db:
        return await db.query(
            """SELECT id, content FROM documents
               WHERE user_id = $1
               ORDER BY embedding NEAR $2
               LIMIT 5""",
            user_id, query_vec
        )

results = asyncio.run(search("u_441", query_embedding))

Type hints and Pydantic models

from datetime import datetime
from pydantic import BaseModel
from dreambase import Vector

class Document(BaseModel):
    id: str
    user_id: str
    content: str
    embedding: Vector[1536]   # typed vector field
    created_at: datetime

# Typed query — returns List[Document]
docs: list[Document] = db.query_as(
    Document,
    "SELECT * FROM documents WHERE user_id=$1 ORDER BY embedding NEAR $2 LIMIT 5",
    user_id, query_vec
)