Python SDK

Python SDK

Package: @heliosdb-sdks family — pip install heliosdb (REST/Realtime client) or pip install psycopg2-binary (direct PG wire) Repo: heliosdb-sdks Compatible with: all HeliosDB editions (Nano, Lite, Full)

---

UVP

Python developers reaching for HeliosDB Nano have two equally valid paths: hit the REST/Realtime BaaS surface with the Supabase-compatible heliosdb package, or talk PostgreSQL wire protocol directly with psycopg2 / asyncpg. This page covers both, with a hands-on snippet for each — connect, CRUD, vector search — and links to the matching end-to-end tutorial. You don’t need a special driver. The Nano binary speaks every Python developer’s existing dialect.

---

Path A — Direct PG-Wire (psycopg2)

Use this when you want full SQL access, transactions, prepared statements, and the smallest possible driver.

Install

pip install psycopg2-binary

Connect

import psycopg2

conn = psycopg2.connect(
    "postgresql://postgres:s3cret@127.0.0.1:5432/postgres"
)
conn.autocommit = True
cur = conn.cursor()

cur.execute("SELECT version()")
print(cur.fetchone()[0])
# PostgreSQL 16.0 (compatible) -- HeliosDB Nano 3.19.1

CRUD

# Schema
cur.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id   SERIAL PRIMARY KEY,
        name TEXT NOT NULL,
        age  INT
    )
""")

# Insert
cur.execute("INSERT INTO users (name, age) VALUES (%s, %s) RETURNING id",
            ("Alice", 30))
new_id = cur.fetchone()[0]

# Read
cur.execute("SELECT id, name, age FROM users WHERE id = %s", (new_id,))
print(cur.fetchone())

# Update
cur.execute("UPDATE users SET age = %s WHERE id = %s", (31, new_id))

# Delete
cur.execute("DELETE FROM users WHERE id = %s", (new_id,))

Vector Search

# Schema with HNSW index
cur.execute("CREATE TABLE IF NOT EXISTS docs ("
            "id SERIAL PRIMARY KEY, title TEXT, embedding VECTOR(384))")
cur.execute("CREATE INDEX IF NOT EXISTS docs_emb_idx "
            "ON docs USING hnsw (embedding vector_cosine_ops)")

# Insert with embedding from sentence-transformers
from sentence_transformers import SentenceTransformer
m = SentenceTransformer("all-MiniLM-L6-v2")

def vec_lit(v):
    return "[" + ",".join(f"{x:.6f}" for x in v) + "]"

emb = m.encode(["Helios Nano release notes"], normalize_embeddings=True)[0]
cur.execute("INSERT INTO docs (title, embedding) VALUES (%s, %s::vector)",
            ("Release notes", vec_lit(emb)))

# k-NN query
q = m.encode(["what's new in helios?"], normalize_embeddings=True)[0]
cur.execute(
    "SELECT id, title, embedding <=> %s::vector AS distance "
    "FROM docs ORDER BY distance LIMIT 5",
    (vec_lit(q),)
)
for row in cur.fetchall():
    print(row)

Async — asyncpg

import asyncio, asyncpg

async def main():
    conn = await asyncpg.connect("postgresql://postgres:s3cret@127.0.0.1:5432/postgres")
    rows = await conn.fetch("SELECT id, name FROM users LIMIT 10")
    for r in rows:
        print(r["id"], r["name"])
    await conn.close()

asyncio.run(main())

---

Path B — REST + Realtime SDK (heliosdb)

Use this when you want the Supabase-style fluent API, automatic JWT handling, and live row subscriptions.

Install

pip install heliosdb

Connect

from heliosdb import create_client

db = create_client("http://localhost:8080", api_key="anon-key")

Auth

# Sign up (also returns a session)
result = db.auth.sign_up({"email": "alice@example.com", "password": "s3cret"})

# Login
session = db.auth.sign_in_with_password({
    "email": "alice@example.com",
    "password": "s3cret",
})
# session.access_token attached to every subsequent request automatically

CRUD over REST

# Insert
post = db.from_("posts").insert({
    "title": "Hello",
    "body":  "First post",
}).execute()

# Read with filter chain (PostgREST-compatible)
rows = (db.from_("posts")
          .select("id, title, created_at")
          .eq("author", "alice@example.com")
          .order("created_at", desc=True)
          .limit(10)
          .execute())

# Update
db.from_("posts").update({"body": "edited"}).eq("id", post.data[0]["id"]).execute()

# Delete
db.from_("posts").delete().eq("id", post.data[0]["id"]).execute()

Vector Search via SDK

results = db.vector_search(
    table="docs",
    query=[0.1, 0.2, 0.3, ...],   # 384-dim vector
    top_k=10,
    metric="cosine",              # "cosine" | "l2" | "ip"
    filter={"category": "tech"},
)
for hit in results:
    print(hit.id, hit.score, hit.row["title"])

Realtime Subscriptions

def on_change(payload):
    print("change:", payload["eventType"], payload["new"])

channel = (db.channel("posts-changes")
             .on_postgres_changes(
                 event="*", schema="public", table="posts",
                 callback=on_change,
             )
             .subscribe())
# channel.unsubscribe() to stop

The SDK opens ws://localhost:8080/realtime/v1/websocket under the hood and emits the same event shape Supabase does.

---

When to Pick Which

Need	Use
Full SQL surface (joins, CTEs, prepared statements, COPY)	Path A — psycopg2 / asyncpg
Multi-tenant app with JWT-driven RLS	Path B — heliosdb SDK
Realtime WebSocket subscriptions	Path B
Bulk loading from Pandas / Parquet	Path A (COPY FROM)
Server-side cursors / large scans	Path A
Drop-in Supabase migration	Path B

You can mix both in the same app — pgx-style CRUD for batch jobs, REST/Realtime for the user-facing API tier.

---

Next Steps

• PYTHON_VECTOR_APP — full semantic search app, end-to-end.
• BAAS_REST_API_TUTORIAL — REST endpoints at the curl level.
• VECTOR_SEARCH_TUTORIAL — distance ops + HNSW tuning.
• TypeScript SDK — same surface in Node.js.