Databases: SQLAlchemy, asyncpg, and connection pools

Why this matters

Almost every Python application touches a database. SQLite ships with Python and runs anywhere; PostgreSQL is the production default; DuckDB is the analytical workhorse; SQLAlchemy is the ORM and connection toolkit you'll see in every Python job. This lesson covers the patterns you actually use.

Learning objectives

Use sqlite3 for local / embedded data.
Use SQLAlchemy 2.0 Core and ORM idiomatically.
Use asyncpg (or SQLAlchemy async) for production PostgreSQL.
Use DuckDB for in-process analytics.
Handle migrations with Alembic.

1. `sqlite3` — built-in, zero install

python

import sqlite3

with sqlite3.connect("app.db") as conn:
    conn.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            email TEXT UNIQUE,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("Ada", "ada@x.com"))
    for row in conn.execute("SELECT * FROM users"):
        print(row)

Use `?` placeholders — never f-strings

python

# BAD — SQL injection
conn.execute(f"SELECT * FROM users WHERE name = '{name}'")

# GOOD
conn.execute("SELECT * FROM users WHERE name = ?", (name,))

Rows as dicts

python

conn.row_factory = sqlite3.Row
for row in conn.execute("SELECT name, email FROM users"):
    print(row["name"], row["email"])

Transactions

By default, sqlite3 opens implicit transactions and commits when you call conn.commit(). Use with conn: to commit on success and rollback on exception:

python

with conn:
    conn.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))
    conn.execute("INSERT INTO users (name) VALUES (?)", ("Cara",))
# both committed or both rolled back

When SQLite is right

Embedded apps, desktop tools, mobile.
Local caches, dev databases, tests.
Single-writer workloads.
Up to ~100 GB; great single-machine performance.

Not right for: many concurrent writers, distributed systems.

2. SQLAlchemy 2.0 — Core

SQLAlchemy is the Swiss-army knife. Two layers: Core (SQL expression language) and ORM (object-relational mapping).

powershell

uv add "sqlalchemy>=2" "psycopg[binary]" "alembic"

python

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, DateTime, select, insert, update, delete
from datetime import datetime

engine = create_engine("postgresql+psycopg://user:pass@localhost/dbname", echo=False)
meta = MetaData()

users = Table(
    "users", meta,
    Column("id", Integer, primary_key=True),
    Column("name", String(50), nullable=False),
    Column("email", String(120), unique=True),
    Column("created_at", DateTime, default=datetime.utcnow),
)

meta.create_all(engine)

with engine.begin() as conn:
    conn.execute(insert(users), [{"name": "Ada", "email": "ada@x.com"}])
    rows = conn.execute(select(users).where(users.c.name.like("A%"))).all()
    for r in rows:
        print(r.id, r.name)

Key Core constructs:

select(table).where(cond)
insert(table).values(...) or conn.execute(insert(table), list_of_dicts)
update(table).where(...).values(...)
delete(table).where(...)
func.count(...), func.now()

engine.begin() opens a transaction. engine.connect() opens without a transaction.

3. SQLAlchemy 2.0 — ORM (the new typed API)

python

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship
from sqlalchemy import ForeignKey, String, DateTime, func

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str | None] = mapped_column(String(120), unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    orders: Mapped[list["Order"]] = relationship(back_populates="user", cascade="all, delete-orphan")

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    amount: Mapped[float]
    user: Mapped[User] = relationship(back_populates="orders")

Base.metadata.create_all(engine)

with Session(engine) as session:
    ada = User(name="Ada", email="ada@x.com")
    ada.orders.append(Order(amount=99.99))
    session.add(ada)
    session.commit()

    rows = session.scalars(select(User).where(User.name == "Ada")).all()
    for u in rows:
        print(u.name, [o.amount for o in u.orders])

Idioms

session.scalars(...) returns the first column (typically the ORM entity).
session.execute(...) returns tuples for ad-hoc selects.
session.add(obj) / session.add_all([...]) + session.commit().
session.flush() writes pending changes without commit.
session.refresh(obj) reloads from DB.

Lazy vs eager loading

python

from sqlalchemy.orm import selectinload, joinedload

# Eager — avoids N+1
stmt = select(User).options(selectinload(User.orders))
for u in session.scalars(stmt):
    print(u.orders)              # already loaded; no extra queries

selectinload issues one extra query per relationship (typically what you want). joinedload joins in the same query (good for one-to-one).

The N+1 problem (one query for users, then N for each user's orders) is the most common SQLAlchemy bug. Always think about loading strategy.

4. Async SQLAlchemy + asyncpg

python

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import select

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/dbname")

async def get_user(uid: int) -> User | None:
    async with AsyncSession(engine) as session:
        return await session.scalar(select(User).where(User.id == uid))

For FastAPI:

python

from fastapi import Depends

async def db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSession(engine) as s:
        yield s

@app.get("/users/{uid}")
async def read_user(uid: int, session: AsyncSession = Depends(db)):
    return await session.scalar(select(User).where(User.id == uid))

Plain `asyncpg` (no ORM)

When raw SQL is preferred (analytical queries, perf-critical paths):

python

import asyncpg

async def main():
    pool = await asyncpg.create_pool("postgresql://user:pass@localhost/db", min_size=2, max_size=10)
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", uid)
        rows = await conn.fetch("SELECT * FROM users")
        await conn.execute("INSERT INTO users (name) VALUES ($1)", "Ada")
    await pool.close()

asyncpg is the fastest PostgreSQL driver — used by many high-throughput services.

5. Connection pools

Always pool. Even for SQLite, but especially for PostgreSQL.

SQLAlchemy: default pool size 5, overflow 10. Tune with pool_size=, max_overflow= on create_engine.
Plain asyncpg: use asyncpg.create_pool.
External pooler (PgBouncer) for thousands of connections.

python

engine = create_engine(
    "postgresql+psycopg://...",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,                   # validate before use
    pool_recycle=3600,                    # recycle every hour
)

pool_pre_ping=True catches "database connection went away" errors gracefully.

6. Migrations with Alembic

powershell

uv add alembic
alembic init migrations

Edit alembic.ini and migrations/env.py to point at your metadata.

powershell

alembic revision --autogenerate -m "add users table"
alembic upgrade head
alembic downgrade -1
alembic history

Always review auto-generated migrations — Alembic gets enums, indexes, and renames wrong sometimes.

Commit migrations to git. Each PR that changes the schema should include a migration.

7. DuckDB — in-process analytics

python

import duckdb

# Query a CSV directly
duckdb.sql("SELECT region, SUM(amount) FROM 'orders.csv' GROUP BY region").df()

# Query Parquet
duckdb.sql("SELECT * FROM 'data/*.parquet' WHERE day = '2026-01-01'").pl()

# Hot interop: query a Polars / pandas DataFrame
import polars as pl
df_pl = pl.read_parquet("daily.parquet")
duckdb.sql("SELECT * FROM df_pl WHERE revenue > 100").pl()

DuckDB is OLAP SQL on a single process. No server. Great for:

ad-hoc SQL on Parquet / CSV / Polars.
ETL where you'd otherwise reach for Spark.
Local analytics in notebooks.

For multi-user / persistent OLTP, use PostgreSQL. For OLAP, DuckDB is increasingly the default.

8. NoSQL — Redis, MongoDB

For caches, queues, sessions: Redis.

python

import redis
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
r.set("user:42", "Ada", ex=3600)               # 1-hour expiry
r.get("user:42")
r.incr("counter:requests")
r.lpush("queue:jobs", json.dumps(job))
job = json.loads(r.rpop("queue:jobs"))

Async: from redis.asyncio import Redis.

For document stores (rare in 2026; usually PG JSONB is enough): motor for async MongoDB.

9. Patterns

Repository pattern with SQLAlchemy

python

class UserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get(self, uid: int) -> User | None:
        return await self.session.scalar(select(User).where(User.id == uid))

    async def by_email(self, email: str) -> User | None:
        return await self.session.scalar(select(User).where(User.email == email))

    async def create(self, name: str, email: str) -> User:
        u = User(name=name, email=email)
        self.session.add(u)
        await self.session.flush()
        return u

Unit of Work

python

async with AsyncSession(engine) as session:
    async with session.begin():
        repo = UserRepository(session)
        await repo.create("Ada", "ada@x.com")
        await repo.create("Bob", "bob@x.com")
    # commit on exit; rollback on exception

Read replicas

For analytics traffic, point a second engine at a read replica. SQLAlchemy supports session binding to multiple engines.

10. Performance tips

Index what you filter / join on. EXPLAIN ANALYZE everything important.
Bulk insert: session.bulk_insert_mappings(User, [...]) or executemany rather than one-by-one.
Stream large reads with stream_results=True + yield_per.
Batch fetches with selectinload to avoid N+1.
Connection pooling: never open a new connection per request.
Prepared statements: free with SQLAlchemy and asyncpg; can be set up explicitly with Connection.prepare().

11. Worked example: a tiny CRUD with FastAPI + SQLAlchemy async

python

# models.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str] = mapped_column(unique=True)

# db.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine(os.getenv("DATABASE_URL"))
async def get_session():
    async with AsyncSession(engine) as session:
        yield session

# api.py
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select

app = FastAPI()

class UserIn(BaseModel): name: str; email: str
class UserOut(BaseModel):
    id: int; name: str; email: str
    model_config = {"from_attributes": True}

@app.post("/users", response_model=UserOut)
async def create(payload: UserIn, session: AsyncSession = Depends(get_session)):
    user = User(**payload.model_dump())
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

@app.get("/users/{uid}", response_model=UserOut)
async def read(uid: int, session: AsyncSession = Depends(get_session)):
    user = await session.scalar(select(User).where(User.id == uid))
    if not user: raise HTTPException(404, "not found")
    return user

A real production stack: FastAPI + Pydantic + SQLAlchemy async + Alembic + asyncpg.

Hands-on lab (2.5 hours)

Create an SQLite database; insert 1000 random rows; query top-10 by a column.
Repeat with SQLAlchemy Core (no ORM). Use insert(...).values(list_of_dicts).
Convert to ORM: define a model, use Session to add/query.
Add a second related table; demonstrate selectinload to avoid N+1.
Set up Alembic; auto-generate a migration; review.
Add an async engine + AsyncSession; build one endpoint with FastAPI.
Try DuckDB on a folder of Parquet files; compare timings to pandas.
Bonus: use Redis to cache the result of an expensive query; observe latency drop.

Common pitfalls

String interpolation in SQL → SQL injection.
N+1 queries — always think about loading.
Forgetting await session.commit() in async code.
Opening + closing a new connection per request — use a pool.
Long-running transactions blocking other writers.
Not handling IntegrityError (unique-constraint violation, FK error).
Using SQLite for high-concurrency production workloads.

Self-check

SQLAlchemy Core vs ORM — when use which?
What is the N+1 query problem? How fix?
Why use a connection pool?
When use SQLite vs PostgreSQL vs DuckDB?
How does Alembic generate migrations?

References

SQLAlchemy 2.0 docs: https://docs.sqlalchemy.org/en/20/.
Essential SQLAlchemy, Rick Copeland.
asyncpg docs: https://magicstack.github.io/asyncpg/.
DuckDB docs: https://duckdb.org/docs/.
Alembic docs: https://alembic.sqlalchemy.org/.
PostgreSQL official docs.

Why this matters

Learning objectives

1. sqlite3 — built-in, zero install

Use ? placeholders — never f-strings

Rows as dicts

Transactions

When SQLite is right

2. SQLAlchemy 2.0 — Core

3. SQLAlchemy 2.0 — ORM (the new typed API)

Idioms

Lazy vs eager loading

4. Async SQLAlchemy + asyncpg

Plain asyncpg (no ORM)

5. Connection pools

6. Migrations with Alembic

7. DuckDB — in-process analytics

8. NoSQL — Redis, MongoDB

9. Patterns

Repository pattern with SQLAlchemy

Unit of Work

Read replicas

10. Performance tips

11. Worked example: a tiny CRUD with FastAPI + SQLAlchemy async

Hands-on lab (2.5 hours)

Common pitfalls

Self-check

References

1. `sqlite3` — built-in, zero install

Use `?` placeholders — never f-strings

Plain `asyncpg` (no ORM)