SQLAlchemy Cheatsheet 01 — Engine, Pool, and Connection

# Postgres
"postgresql+asyncpg://user:pass@host/db"      # async (recommended)
"postgresql+psycopg://user:pass@host/db"      # psycopg v3 (sync or async)
"postgresql+psycopg2://user:pass@host/db"     # legacy sync

# MySQL
"mysql+aiomysql://user:pass@host/db"
"mysql+asyncmy://user:pass@host/db"
"mysql+pymysql://user:pass@host/db"

# SQLite
"sqlite:///app.db"
"sqlite+aiosqlite:///app.db"
"sqlite:///:memory:"

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_engine(URL, echo=False)
engine = create_async_engine(URL, echo=False)

engine = create_async_engine(
    URL,
    pool_size=20,            # persistent connections
    max_overflow=10,         # extra under load
    pool_timeout=30,         # wait this long for free conn
    pool_pre_ping=True,      # ping on checkout (catch dead conns)
    pool_recycle=3600,       # recycle conns > N seconds
    echo=False,
    echo_pool=False,
    future=True,
)

from sqlalchemy.pool import NullPool, StaticPool, QueuePool

# No pool (serverless)
engine = create_engine(URL, poolclass=NullPool)

# Single shared connection (SQLite in-memory tests)
engine = create_engine(
    "sqlite://",
    poolclass=StaticPool,
    connect_args={"check_same_thread": False},
)

# asyncpg
engine = create_async_engine(URL, connect_args={
    "server_settings": {
        "application_name": "myapp",
        "search_path": "myapp,public",
        "timezone": "UTC",
    },
    "command_timeout": 60,
    "statement_cache_size": 1000,
})

# psycopg v3
engine = create_async_engine(URL, connect_args={
    "options": "-c search_path=myapp,public -c application_name=myapp",
})

# Sync
with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))
    print(result.scalar())

with engine.begin() as conn:    # auto-commit on exit
    conn.execute(text("CREATE TABLE ..."))

# Async
async with engine.connect() as conn:
    result = await conn.execute(text("SELECT 1"))

async with engine.begin() as conn:
    await conn.execute(text("..."))

print(engine.pool.status())
print(engine.pool.size(), engine.pool.checked_out())

from sqlalchemy import event

@event.listens_for(engine.sync_engine, "connect")
def on_connect(dbapi_conn, conn_record):
    cur = dbapi_conn.cursor()
    cur.execute("SET TIME ZONE 'UTC'")
    cur.execute("SET statement_timeout = '30s'")
    cur.close()

@event.listens_for(engine.sync_engine, "checkout")
def on_checkout(dbapi_conn, conn_record, conn_proxy):
    ...

import time

@event.listens_for(engine.sync_engine, "before_cursor_execute")
def before(conn, cur, stmt, params, ctx, executemany):
    ctx._t = time.time()

@event.listens_for(engine.sync_engine, "after_cursor_execute")
def after(conn, cur, stmt, params, ctx, executemany):
    dur = time.time() - ctx._t
    if dur > 0.5:
        log.warning("slow_query", sql=stmt[:500], ms=dur*1000)

@asynccontextmanager
async def lifespan(app):
    app.state.engine = create_async_engine(URL, pool_size=20, pool_pre_ping=True)
    app.state.sm = async_sessionmaker(app.state.engine, expire_on_commit=False)
    yield
    await app.state.engine.dispose()

async def get_db(request: Request) -> AsyncSession:
    async with request.app.state.sm() as session:
        yield session

write_engine = create_async_engine(WRITER_URL, pool_size=20)
read_engine = create_async_engine(
    READER_URL,
    pool_size=40,
    connect_args={"server_settings": {"default_transaction_read_only": "on"}},
)

await engine.dispose()   # close all pooled connections