Use Case Examples

Real-world patterns with HeliosDB-Lite

Every example runs SQL against heliosdb-lite start. Statements that change data or schema go to POST /v1/branches/main/execute, and queries go to POST /v1/branches/main/query. You can run the same SQL over the PostgreSQL wire with any PostgreSQL driver, or in embedded mode. The REST vector-store, document, agent-memory and chat endpoints are planned, so these examples store everything in ordinary tables and use SQL vector search.

---

Quick Links

Use Case	Description	Complexity
RAG Backend	Retrieval-Augmented Generation	Medium
AI Agent Memory	Persistent conversation context	Easy
Multi-Tenant SaaS	Row-level security per tenant	Medium
Semantic Search	Product and document similarity	Easy
Event Analytics	User behavior tracking	Medium
Feature Flags	Dynamic configuration	Easy

---

Shared REST Helper (Python)

The examples below use this small client. TOKEN is a JWT signed with HELIOSDB_JWT_SECRET; see Authentication.

import requests

class HeliosLite:
    def __init__(self, base_url: str, token: str, branch: str = "main"):
        self.base = f"{base_url}/v1/branches/{branch}"
        self.headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

    def execute(self, sql: str, params: list | None = None) -> dict:
        r = requests.post(f"{self.base}/execute", headers=self.headers,
                          json={"sql": sql, "params": params or []})
        r.raise_for_status()
        return r.json()

    def query(self, sql: str, params: list | None = None) -> list[dict]:
        r = requests.post(f"{self.base}/query", headers=self.headers,
                          json={"sql": sql, "params": params or []})
        r.raise_for_status()
        return r.json()["rows"]

def p(value):
    """Wrap a Python value as a typed REST parameter."""
    if value is None:
        return {"type": "null"}
    if isinstance(value, bool):
        return {"type": "boolean", "value": value}
    if isinstance(value, int):
        return {"type": "int8", "value": value}
    if isinstance(value, float):
        return {"type": "float8", "value": value}
    if isinstance(value, (dict, list)):
        return {"type": "json", "value": value}
    return {"type": "string", "value": str(value)}

def vec(values) -> str:
    """Render an embedding as a SQL vector literal (numbers only)."""
    return "'[" + ", ".join(f"{float(x):.6f}" for x in values) + "]'"

---

RAG Backend

Use HeliosDB-Lite as the vector store for Retrieval-Augmented Generation.

Architecture

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│  Documents  │────▶│ Embeddings   │────▶│ HeliosDB    │
│  (PDF, MD)  │     │ (your model) │     │ Lite (HNSW) │
└─────────────┘     └──────────────┘     └─────────────┘
                                               │
┌─────────────┐     ┌──────────────┐           │
│  User Query │────▶│     LLM      │◀──────────┘
└─────────────┘     └──────────────┘     Retrieved Context

Setup

CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    source TEXT NOT NULL,
    chunk_index INTEGER NOT NULL,
    content TEXT NOT NULL,
    embedding VECTOR(1536) NOT NULL,
    created_at TIMESTAMP DEFAULT now()
);

-- HNSW index; add quantization = 'product' to store compact codes instead of full vectors
CREATE INDEX documents_embedding_idx ON documents
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 200);

Python Implementation

from openai import OpenAI

class RAGBackend:
    def __init__(self, db: HeliosLite, client: OpenAI):
        self.db = db
        self.client = client

    def embed(self, text: str) -> list[float]:
        return self.client.embeddings.create(
            model="text-embedding-3-small", input=text
        ).data[0].embedding

    def ingest_document(self, source: str, content: str, chunk_words: int = 500):
        words = content.split()
        for idx in range(0, len(words), chunk_words):
            chunk = " ".join(words[idx:idx + chunk_words])
            self.db.execute(
                "INSERT INTO documents (source, chunk_index, content, embedding) "
                f"VALUES ($1, $2, $3, {vec(self.embed(chunk))})",
                [p(source), p(idx // chunk_words), p(chunk)],
            )

    def search(self, question: str, limit: int = 5) -> list[dict]:
        q = vec(self.embed(question))
        return self.db.query(
            f"SELECT source, content, embedding <=> {q} AS distance "
            f"FROM documents ORDER BY embedding <=> {q} LIMIT {int(limit)}"
        )

    def answer(self, question: str) -> str:
        context = "\n\n".join(row["content"] for row in self.search(question))
        reply = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": f"Answer from this context:\n{context}"},
                {"role": "user", "content": question},
            ],
        )
        return reply.choices[0].message.content


rag = RAGBackend(HeliosLite("http://127.0.0.1:8080", TOKEN), OpenAI())
rag.ingest_document("docs/readme.md", open("README.md").read())
print(rag.answer("How do I set up authentication?"))

---

AI Agent Memory

Persist agent conversations in ordinary tables. Dedicated REST agent-memory endpoints are planned.

Setup

CREATE TABLE agent_sessions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    agent_id TEXT NOT NULL,
    user_id TEXT,
    started_at TIMESTAMP DEFAULT now(),
    summary TEXT
);

CREATE TABLE agent_messages (
    id SERIAL PRIMARY KEY,
    session_id UUID REFERENCES agent_sessions(id),
    role TEXT NOT NULL,
    content TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT now()
);

CREATE INDEX idx_messages_session ON agent_messages (session_id, created_at);

Python Implementation

class AgentMemory:
    def __init__(self, db: HeliosLite):
        self.db = db

    def create_session(self, agent_id: str, user_id: str | None = None) -> str:
        session_id = self.db.query("SELECT gen_random_uuid() AS id")[0]["id"]
        self.db.execute(
            "INSERT INTO agent_sessions (id, agent_id, user_id) VALUES ($1, $2, $3)",
            [p(session_id), p(agent_id), p(user_id)],
        )
        return session_id

    def add_message(self, session_id: str, role: str, content: str):
        self.db.execute(
            "INSERT INTO agent_messages (session_id, role, content) VALUES ($1, $2, $3)",
            [p(session_id), p(role), p(content)],
        )

    def history(self, session_id: str, limit: int = 50) -> list[dict]:
        return self.db.query(
            "SELECT role, content FROM agent_messages WHERE session_id = $1 "
            f"ORDER BY created_at LIMIT {int(limit)}",
            [p(session_id)],
        )


memory = AgentMemory(HeliosLite("http://127.0.0.1:8080", TOKEN))
sid = memory.create_session("support-bot", "user-123")
memory.add_message(sid, "user", "How do I reset my password?")
print(memory.history(sid))

---

Multi-Tenant SaaS

Give each tenant its own database role and let row-level security filter rows by current_user. A REST request runs as the role in its token’s sub claim, so the same policy protects REST and PostgreSQL wire clients.

Setup (run as a superuser over the PostgreSQL wire)

CREATE TABLE projects (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant TEXT NOT NULL,
    name TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT now()
);

CREATE POLICY projects_tenant_isolation ON projects FOR ALL USING (tenant = current_user);
ALTER TABLE projects ENABLE ROW LEVEL SECURITY;

-- One role per tenant
CREATE ROLE acme LOGIN PASSWORD 'acme-secret';
CREATE ROLE startup LOGIN PASSWORD 'startup-secret';
GRANT SELECT, INSERT, UPDATE, DELETE ON projects TO acme;
GRANT SELECT, INSERT, UPDATE, DELETE ON projects TO startup;

Implementation

Mint one token per tenant role with sub set to the role name (see Authentication):

acme = HeliosLite("http://127.0.0.1:8080", mint_token(sub="acme"))
startup = HeliosLite("http://127.0.0.1:8080", mint_token(sub="startup"))

acme.execute("INSERT INTO projects (tenant, name) VALUES (current_user, 'Acme Project')")
startup.execute("INSERT INTO projects (tenant, name) VALUES (current_user, 'Startup MVP')")

print(acme.query("SELECT name FROM projects"))     # only Acme's rows
print(startup.query("SELECT name FROM projects"))  # only Startup's rows

Superusers (by default postgres and helios) bypass row-level security, so never mint tenant tokens for them. On the PostgreSQL wire, start the server with --auth scram-sha-256 so that clients cannot claim another tenant’s role.

---

Semantic Search

Search products by meaning with a local embedding model.

Setup

CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    description TEXT,
    category TEXT,
    price DECIMAL(10,2),
    embedding VECTOR(384)  -- all-MiniLM-L6-v2 dimension
);

CREATE INDEX products_embedding_idx ON products
    USING hnsw (embedding vector_cosine_ops);

Implementation

from sentence_transformers import SentenceTransformer

class SemanticSearch:
    def __init__(self, db: HeliosLite):
        self.db = db
        self.model = SentenceTransformer("all-MiniLM-L6-v2")

    def index_product(self, name, description, category, price):
        emb = vec(self.model.encode(f"{name}. {description}. Category: {category}").tolist())
        self.db.execute(
            "INSERT INTO products (name, description, category, price, embedding) "
            f"VALUES ($1, $2, $3, $4, {emb})",
            [p(name), p(description), p(category), p(price)],
        )

    def search(self, text, category=None, max_price=None, limit=10):
        q = vec(self.model.encode(text).tolist())
        filters, params = [], []
        if category:
            params.append(p(category)); filters.append(f"category = ${len(params)}")
        if max_price is not None:
            params.append(p(float(max_price))); filters.append(f"price <= ${len(params)}")
        where = f"WHERE {' AND '.join(filters)}" if filters else ""
        return self.db.query(
            f"SELECT name, price, embedding <=> {q} AS distance FROM products {where} "
            f"ORDER BY embedding <=> {q} LIMIT {int(limit)}",
            params,
        )


search = SemanticSearch(HeliosLite("http://127.0.0.1:8080", TOKEN))
search.index_product("Wireless Noise-Canceling Headphones",
                     "Over-ear headphones with ANC and 30-hour battery", "electronics", 299.99)
print(search.search("headphones for working from home", category="electronics", max_price=350))

---

Event Analytics

Track user behavior and aggregate it with SQL.

Setup

CREATE TABLE events (
    id BIGSERIAL PRIMARY KEY,
    user_id TEXT NOT NULL,
    event_type TEXT NOT NULL,
    properties JSONB,
    occurred_at TIMESTAMP DEFAULT now()
);

CREATE INDEX events_user_time_idx ON events (user_id, occurred_at);
CREATE INDEX events_type_time_idx ON events (event_type, occurred_at);

Implementation

class EventTracker:
    def __init__(self, db: HeliosLite):
        self.db = db

    def track(self, user_id: str, event_type: str, properties: dict | None = None):
        self.db.execute(
            "INSERT INTO events (user_id, event_type, properties) VALUES ($1, $2, $3)",
            [p(user_id), p(event_type), p(properties or {})],
        )

    def daily_counts(self, event_type: str) -> list[dict]:
        return self.db.query(
            "SELECT date_trunc('day', occurred_at) AS day, "
            "COUNT(*) AS events, COUNT(DISTINCT user_id) AS users "
            "FROM events WHERE event_type = $1 GROUP BY day ORDER BY day DESC",
            [p(event_type)],
        )


tracker = EventTracker(HeliosLite("http://127.0.0.1:8080", TOKEN))
tracker.track("user-123", "page_view", {"page": "/products"})
tracker.track("user-123", "purchase", {"order_id": "order-789", "total": 29.99})
print(tracker.daily_counts("purchase"))

---

Feature Flags

Keep feature flags in a table and evaluate them in the application.

Setup

CREATE TABLE feature_flags (
    id TEXT PRIMARY KEY,
    enabled BOOLEAN DEFAULT false,
    rollout_percentage INTEGER DEFAULT 0,
    target_plans TEXT,          -- comma-separated plan names, or NULL for all plans
    updated_at TIMESTAMP DEFAULT now()
);

INSERT INTO feature_flags (id, enabled, rollout_percentage, target_plans) VALUES
    ('new_checkout', true, 50, 'pro,enterprise'),
    ('dark_mode', true, 100, NULL),
    ('beta_features', false, 0, 'enterprise');

Implementation

import hashlib

class FeatureFlags:
    def __init__(self, db: HeliosLite):
        self.db = db

    def is_enabled(self, flag_id: str, user_id: str, plan: str | None = None) -> bool:
        rows = self.db.query("SELECT * FROM feature_flags WHERE id = $1", [p(flag_id)])
        if not rows or not rows[0]["enabled"]:
            return False
        flag = rows[0]
        if flag["target_plans"] and plan not in flag["target_plans"].split(","):
            return False
        bucket = int(hashlib.sha256(f"{flag_id}:{user_id}".encode()).hexdigest()[:8], 16) % 100
        return bucket < flag["rollout_percentage"]


flags = FeatureFlags(HeliosLite("http://127.0.0.1:8080", TOKEN))
print(flags.is_enabled("new_checkout", "user-123", "pro"))

---

Next Steps

• API Reference
• Authentication Guide
• Schema Generation

---

Need help? Join Discord | Email support