Using SQLAlchemy with Supabase Postgres

SQLAlchemy connects to Postgres directly. On SupaHost, direct connections go through an SSH tunnel on the Business and Enterprise plans. On Builder and Pro, use supabase-py and the Data API from Python instead; it works on every plan and goes through your RLS policies.

Connect over the tunnel

1. Add your SSH public key: Dashboard → your project → Connect tab → SSH & CLI → Add key.
2. Open the tunnel and keep it running:

ssh -N -L 5432:127.0.0.1:5432 [email protected]

3. Copy the connection string from Connect → Database → Show connection string. It looks like postgresql://postgres:<password>@127.0.0.1:5432/p_app_one. The database is p_ plus the project name with dashes as underscores.

4. Use it with a driver prefix. With psycopg 3 (pip install "sqlalchemy>=2" "psycopg[binary]"):

import os
from sqlalchemy import create_engine, text

engine = create_engine(
    os.environ["DATABASE_URL"].replace("postgresql://", "postgresql+psycopg://", 1),
    pool_size=5,
    max_overflow=2,
    pool_pre_ping=True,   # replaces connections that died when the tunnel dropped
    pool_recycle=1800,
)

with engine.connect() as conn:
    print(conn.execute(text("select current_database(), version()")).one())

There's no connection pooler on SupaHost, so you connect straight to Postgres. You don't need NullPool, prepare_threshold=None or other workarounds for transaction-mode poolers; SQLAlchemy's own pool is the right choice.

Async with asyncpg

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://postgres:[email protected]:5432/p_app_one",
    pool_size=5,
    max_overflow=2,
    pool_pre_ping=True,
)

asyncpg's prepared statement cache works normally here, for the same reason.

Size the pool for a shared server

Every project on your instance shares one Postgres server with 400 connections in total. A web app with 4 workers and pool_size=5, max_overflow=2 can hold up to 28. Add up all your processes (web workers, background workers, notebooks, Alembic) and keep well under the limit. If you hit it, see Too many connections.

Models that reference Supabase users

Keep your tables in public and refer to Auth users by id. A minimal stand-in for auth.users in the same metadata lets SQLAlchemy resolve the foreign key without managing that table:

import uuid
from sqlalchemy import Column, ForeignKey, String, Table
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

# stand-in only: Auth owns this table, never create or alter it
Table("users", Base.metadata, Column("id", UUID(as_uuid=True), primary_key=True), schema="auth")

class Expense(Base):
    __tablename__ = "expenses"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("auth.users.id", ondelete="CASCADE"))
    description: Mapped[str] = mapped_column(String(200))

Don't call Base.metadata.create_all() against the project; use migrations.

Alembic: leave Supabase's schemas alone

Tables in auth, storage and the other Supabase schemas belong to Auth, Storage and the platform. Tell Alembic's autogenerate to ignore them, both the stand-in above and anything it reflects, in env.py:

MANAGED = {"auth", "storage", "realtime", "extensions", "graphql", "graphql_public",
           "pgmq", "pgmq_public", "supabase_functions", "supabase_migrations", "vault", "cron", "net"}

def include_object(obj, name, type_, reflected, compare_to):
    schema = getattr(obj, "schema", None) or getattr(getattr(obj, "table", None), "schema", None)
    return schema not in MANAGED

context.configure(connection=connection, target_metadata=target_metadata,
                  include_object=include_object)

Leave include_schemas at its default (False) so Alembic only compares public.

RLS doesn't apply to this connection

The tunnel connects as postgres, which bypasses RLS in Supabase's Postgres image. Your policies won't filter SQLAlchemy queries, so check permissions in your Python code. Use supabase-py with a user's token when you want RLS to apply.

Related

• Use SupaHost with Python
• SSH access and direct Postgres
• Connecting directly to Postgres (psql, pg_dump, ORMs)
• Connections and timeouts