SQLAlchemy

SQLAlchemy connects to DB9 using the psycopg (psycopg3) driver over pgwire. The modern SQLAlchemy 2.0 API with mapped_column, type-annotated models, and session-based transactions works without modification.

DB9’s E2E test suite validates CRUD, JSONB, joins, subqueries, CTEs, window functions, DDL, transactions, savepoints, and vector operations through SQLAlchemy.

Prerequisites

• A DB9 database (create one)
• Python 3.10+
• SQLAlchemy 2.0+
• psycopg 3.1+ (with binary support)

Create a DB9 Database

Terminal

db9 create --name sqlalchemy-app

db9 create prints the connection string, with the admin password, once — copy it from the Connection String: line. It has this shape:

Output

postgresql://YOUR_TENANT_ID.admin:YOUR_PASSWORD@pg.d0000.db9.io:5432/postgres

YOUR_TENANT_ID is the tenant ID prefix in the generated connection string. Copy the username from the Connection String: line printed by db9 create or from db9 db connect. The examples below use pg.db9.io:5433, which reaches the same server; either host and port works. db9 db status sqlalchemy-app shows the connection string without the password; if you lose it, db9 db reset-password sqlalchemy-app sets a new password and prints a new connection string.

Project Setup

Terminal

mkdir sqlalchemy-db9 && cd sqlalchemy-db9
python -m venv .venv && source .venv/bin/activate
pip install "sqlalchemy>=2.0.0" "psycopg[binary]>=3.1.0"

For vector search, also install:

Terminal

pip install "pgvector>=0.2.5"

Connection

SQLAlchemy requires the postgresql+psycopg:// prefix. Normalize the connection string from DB9:

db.py

from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session, sessionmaker

def build_engine(database_url: str) -> Engine:
    url = database_url.strip()
    if url.startswith("postgresql://"):
        url = "postgresql+psycopg://" + url[len("postgresql://"):]
    elif url.startswith("postgres://"):
        url = "postgresql+psycopg://" + url[len("postgres://"):]
    return create_engine(url, pool_pre_ping=True)

engine = build_engine(
    "postgresql://YOUR_TENANT_ID.admin:YOUR_PASSWORD@pg.db9.io:5433/postgres"
)
SessionFactory = sessionmaker(bind=engine, expire_on_commit=False)

Key engine options:

• pool_pre_ping=True verifies the connection before each use.
• expire_on_commit=False keeps loaded attributes accessible after commit.

Define Models

Use SQLAlchemy 2.0 declarative style with mapped_column:

models.py

from datetime import datetime
from sqlalchemy import Integer, String, Text, DateTime, ForeignKey, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    metadata_: Mapped[dict] = mapped_column("metadata", JSONB, default=dict)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    posts: Mapped[list["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(500), nullable=False)
    content: Mapped[str | None] = mapped_column(Text)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )
    author: Mapped["User"] = relationship(back_populates="posts")

Create Tables

Use Base.metadata.create_all to create tables from your model definitions:

setup.py

from db import engine
from models import Base

Base.metadata.create_all(engine)
print("Tables created")

Base.metadata.create_all(engine) above works against DB9. For later changes, Alembic’s revision --autogenerate works once its comparison is limited to your models’ tables — see Alembic autogenerate needs a table filter. You can also apply a change directly:

Python

from sqlalchemy import text

with engine.begin() as conn:
    conn.execute(text("ALTER TABLE users ADD COLUMN IF NOT EXISTS bio TEXT"))
    conn.execute(text("CREATE INDEX IF NOT EXISTS idx_users_email ON users (email)"))

CRUD Operations

Python

from sqlalchemy import select, text
from db import SessionFactory
from models import User, Post

# Create
with SessionFactory() as session:
    user = User(email="alice@example.com", name="Alice", metadata_={"role": "admin"})
    session.add(user)
    session.commit()

    post = Post(title="Getting Started", content="Hello DB9", author_id=user.id)
    session.add(post)
    session.commit()

# Read
with SessionFactory() as session:
    user = session.get(User, 1)
    print(user.name)  # "Alice"

    users = session.execute(
        select(User).where(User.name == "Alice")
    ).scalars().all()

# Update
with SessionFactory() as session:
    user = session.get(User, 1)
    user.name = "Alice Updated"
    session.commit()

# Upsert (INSERT ... ON CONFLICT)
from sqlalchemy.dialects.postgresql import insert

with SessionFactory() as session:
    stmt = insert(User).values(email="alice@example.com", name="Alice V2")
    stmt = stmt.on_conflict_do_update(
        index_elements=["email"],
        set_={"name": stmt.excluded.name},
    )
    session.execute(stmt)
    session.commit()

# Delete
with SessionFactory() as session:
    user = session.get(User, 1)
    session.delete(user)  # cascades to posts
    session.commit()

JSONB Queries

SQLAlchemy’s PostgreSQL JSONB support works with DB9:

Python

from sqlalchemy import select
from models import User

with SessionFactory() as session:
    # Containment query
    admins = session.execute(
        select(User).where(User.metadata_.contains({"role": "admin"}))
    ).scalars().all()

# Raw SQL for advanced JSONB operations
from sqlalchemy import text

with engine.begin() as conn:
    conn.execute(text(
        "UPDATE users SET metadata = jsonb_set(metadata, '{level}', '5'::jsonb) WHERE id = 1"
    ))

Transactions

Python

# Explicit commit
with SessionFactory() as session:
    session.add(User(email="bob@example.com", name="Bob"))
    session.flush()  # sends to DB but doesn't commit
    session.commit()  # commits the transaction

# Rollback
with SessionFactory() as session:
    session.add(User(email="temp@example.com", name="Temp"))
    session.flush()
    session.rollback()
    # user was not persisted

# Savepoints
with engine.begin() as conn:
    conn.execute(text("SAVEPOINT sp1"))
    conn.execute(text("INSERT INTO users (email, name) VALUES ('x@x.com', 'X')"))
    conn.execute(text("ROLLBACK TO SAVEPOINT sp1"))
    conn.execute(text("RELEASE SAVEPOINT sp1"))

Supported isolation levels: READ COMMITTED, REPEATABLE READ. SERIALIZABLE is not implemented — it does not error, it is silently downgraded to REPEATABLE READ with a server WARNING. Set REPEATABLE READ explicitly so the intent is visible in your code.

Python

with engine.begin() as conn:
    conn.execute(text("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ"))
    # ... queries run at the requested isolation level

Joins and Aggregations

Python

from sqlalchemy import select, func
from models import User, Post

with SessionFactory() as session:
    # Join with aggregation
    results = session.execute(
        select(User.name, func.count(Post.id).label("post_count"))
        .join(Post, User.id == Post.author_id)
        .group_by(User.name)
        .having(func.count(Post.id) >= 1)
        .order_by(User.name)
    ).all()

    # Subquery
    subq = (
        select(Post.author_id, func.count().label("cnt"))
        .group_by(Post.author_id)
        .subquery()
    )
    top_authors = session.execute(
        select(User.name)
        .join(subq, User.id == subq.c.author_id)
        .where(subq.c.cnt >= 2)
    ).scalars().all()

Window Functions and CTEs

Use raw SQL for window functions and CTEs:

Python

from sqlalchemy import text

with engine.begin() as conn:
    # Window function
    rows = conn.execute(text("""
        SELECT name, email,
            ROW_NUMBER() OVER (ORDER BY created_at) AS row_num
        FROM users
    """)).fetchall()

    # CTE
    rows = conn.execute(text("""
        WITH recent AS (
            SELECT * FROM posts WHERE created_at > now() - interval '7 days'
        )
        SELECT u.name, r.title
        FROM users u JOIN recent r ON u.id = r.author_id
    """)).fetchall()

Vector Search

Install pgvector for SQLAlchemy vector type support:

vector_models.py

from pgvector.sqlalchemy import Vector
from sqlalchemy import Integer, Text
from sqlalchemy.orm import Mapped, mapped_column
from models import Base

EMBEDDING_DIM = 1024

class Document(Base):
    __tablename__ = "documents"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    content: Mapped[str] = mapped_column(Text, nullable=False)
    embedding: Mapped[list[float]] = mapped_column(Vector(EMBEDDING_DIM), nullable=False)

Create the table and HNSW index:

Python

from sqlalchemy import text
from db import engine
from vector_models import Base

with engine.begin() as conn:
    conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
    Base.metadata.create_all(conn)
    conn.execute(text("""
        CREATE INDEX IF NOT EXISTS idx_documents_embedding
        ON documents USING hnsw (embedding vector_cosine_ops)
        WITH (m = 16, ef_construction = 64)
    """))

Search by cosine distance:

Python

from sqlalchemy import select
from vector_models import Document

def search(session, query_embedding: list[float], top_k: int = 5):
    distance = Document.embedding.cosine_distance(query_embedding).label("distance")
    stmt = (
        select(Document, distance)
        .order_by(distance)
        .limit(top_k)
    )
    return session.execute(stmt).all()

with SessionFactory() as session:
    results = search(session, query_embedding=[0.1, 0.2, ...])
    for doc, dist in results:
        print(f"{doc.content} (distance: {dist:.4f})")

IVFFlat indexes are not supported

DB9 supports HNSW indexes (vector_cosine_ops, vector_l2_ops, vector_ip_ops) but not IVFFlat. If your code uses CREATE INDEX ... USING ivfflat, change it to USING hnsw — HNSW index building works over both the wire protocol and the HTTP SQL API. SQLAlchemy binds the query vector as a parameter, and a bound probe uses the index when the query also has a LIMIT and no WHERE clause (see When the index is used). Keep the .label("distance") on the expression you pass to order_by(), as search() does: SQLAlchemy then renders ORDER BY distance. Ordering by the unlabelled expression repeats the uncast parameter in the select list and the ORDER BY, and that plan reads every row.

The supported operator classes are vector_cosine_ops, vector_l2_ops and vector_ip_ops; IVFFlat is not available at all.

Full-Text Search

DB9 supports PostgreSQL full-text search with to_tsvector, websearch_to_tsquery, and ts_rank:

Python

from sqlalchemy import text

with SessionFactory() as session:
    results = session.execute(text("""
        SELECT content,
            ts_rank(to_tsvector('english', content),
                    websearch_to_tsquery('english', :query)) AS rank
        FROM documents
        WHERE to_tsvector('english', content) @@ websearch_to_tsquery('english', :query)
        ORDER BY rank DESC
        LIMIT :top_k
    """), {"query": "database serverless", "top_k": 10}).mappings().all()

Index the to_tsvector(...) expression, not the text column. A GIN index over a bare TEXT column is rejected with data type text has no default operator class for access method "gin" (42704) — only TSVECTOR, JSONB and array types have a GIN operator class, so USING GIN (content) never builds:

Python

from sqlalchemy import text
from db import engine

with engine.begin() as conn:
    conn.execute(text("""
        CREATE INDEX IF NOT EXISTS idx_documents_fts
        ON documents USING GIN (to_tsvector('english', content))
    """))

The expression in the index must match the one in the predicate. The query shown earlier binds the search term as a parameter, and that uses the index: a bound term inside websearch_to_tsquery('english', :query) is indexed just as a literal is. Keep the configuration ('english') written in the query — a call that omits it or binds it as a parameter plans as a Seq Scan with KV Table Scan Pairs equal to the row count, identical to having no index at all. HNSW’s extra conditions do not apply here: a WHERE clause and the absence of a LIMIT are both fine. Confirm with EXPLAIN ANALYZE rather than assuming the index is in play.

Production Notes

• Driver: Use psycopg (psycopg3), with the connection string prefix postgresql+psycopg://.
• Connection pooling: SQLAlchemy’s built-in pool works with DB9. Use pool_pre_ping=True to handle idle connection drops.
• Alembic migrations: upgrade, downgrade, current, the version table and revision --autogenerate all work against DB9 — autogenerate once the comparison is limited to your models’ tables (see Troubleshooting).
• hstore: DB9 does not implement hstore values — the extension is a metadata-only shim, and any use of the type fails with 0A000 — so map key-value data to JSONB (sqlalchemy.dialects.postgresql.JSONB).
• Connection string: Use the format postgresql+psycopg://<tenant_id>.admin:<password>@pg.db9.io:5433/postgres, using the tenant ID prefix from the generated connection string.

Troubleshooting

OperationalError: FATAL: authentication failed

Verify the tenant-qualified username. DB9 expects <tenant_id>.admin; use the complete username from the generated connection string or db9 db connect. sqlalchemy-app.admin fails even with the right password.

psycopg2 import errors

This guide’s code uses psycopg (v3), not psycopg2. Install psycopg[binary] and use the postgresql+psycopg:// prefix for the application engine. (psycopg2 itself also works against DB9, including for Alembic.)

CREATE EXTENSION vector fails

DB9 includes pgvector natively. The extension may already be available. Wrap the call in a try/except or use CREATE EXTENSION IF NOT EXISTS vector.

Alembic autogenerate needs a table filter

alembic revision --autogenerate runs on both the psycopg (v3) driver this guide uses and psycopg2, but needs one extra step. DB9’s pg_table_is_visible() returns true for every table in every schema, whatever the search_path — including the three tables of the pre-installed pg_cron extension (job, job_run_details and running_jobs in the cron schema) and any table in a schema of your own. Alembic takes them all for stray tables in the default schema, writes an op.drop_table(...) for each into every generated migration, and alembic upgrade then fails on whichever it reaches first — relation "job" does not exist, for example — without applying anything. Restrict the comparison to the tables your models define, in migrations/env.py:

Python

def include_name(name, type_, parent_names):
    return type_ != "table" or name in target_metadata.tables


# in run_migrations_online():
context.configure(
    connection=connection,
    target_metadata=target_metadata,
    include_name=include_name,
)

With that in place autogenerate detects new tables and added columns, and produces an empty migration when nothing changed. The cost is that deleting a model no longer generates a drop_table; write that one by hand. The filter is also by name only: a table in another schema with the same name as one of your models is reflected together with it, and autogenerate then proposes drop_column and drop_index operations for the other schema’s columns and indexes. Avoid the duplicate name, or delete those operations from the generated migration.

alembic upgrade head, alembic downgrade and alembic current behave normally, and the alembic_version table is maintained for you.

Reflection works on both drivers: the inspector methods get_table_names(), get_columns(), get_pk_constraint(), get_foreign_keys(), get_indexes(), get_unique_constraints(), get_enums() and get_domains(), and with them Table(..., autoload_with=engine) and MetaData.reflect(). For the same visibility reason, get_table_names() lists job, job_run_details and running_jobs — and tables from your other schemas — next to the default schema’s own tables, and an unfiltered MetaData.reflect() brings them in too: pass only= to leave them out.

Verified Compatibility

Tested with SQLAlchemy 2.0+ and psycopg 3.1+ against DB9. E2E smoke tests cover:

Category	Status
Connection and pooling	Pass
CRUD operations	Pass
JSONB containment queries	Pass
Transactions and savepoints	Pass
Joins and aggregations	Pass
Subqueries and CTEs	Pass
Window functions	Pass
DDL (ALTER TABLE, CREATE INDEX)	Pass
Vector operations and HNSW indexing	Pass
Upsert (ON CONFLICT)	Pass

Next Pages

• RAG with Built-in Embeddings — vector search with DB9-native embedding
• Connect — connection strings and authentication
• Vector Extension — HNSW indexes and distance operators
• Compatibility Matrix — full PostgreSQL compatibility surface
• Prisma — Node.js ORM alternative
• Drizzle — TypeScript ORM alternative