SQLAlchemy 2.0 Textbook Ch. 1 — Introduction, Core vs ORM, and the 2.0 Mental Model

This is Chapter 1 of the SQLAlchemy 2.0 textbook (database-agnostic). For a Postgres-specific companion, see the Postgres-focused textbook .

We start with SQLAlchemy 2.0’s mental model: what Core is, what ORM is, how they unified, and the basic engine/connection lifecycle. Notes on Postgres / MySQL / SQLite differences scattered throughout.

What SQLAlchemy is

SQLAlchemy is a SQL toolkit and ORM. Two layers, same library:

Core: SQL expressions, schema definitions, connection pooling, dialects. Like a typed query builder + connection layer.
ORM: Object-relational mapping built on Core. Sessions, units of work, relationships, identity map.

They’re not separate libraries; they’re layers. ORM uses Core under the hood. You can use Core alone (without the ORM) but not the other way around.

SQLAlchemy 2.0 changes

The 2.0 release (and the 1.4 transition) unified the API:

One way to write queries: select(...) (the future-style).
One way to execute: session.execute(...) returns a Result.
Strict typing throughout.
Async support throughout.
Mapped attributes typed via Mapped[...].
New declarative base via DeclarativeBase.

If you’re reading old tutorials with Query.filter_by(...): that’s legacy 1.x. Stick to 2.0 patterns.

Engine and dialects

from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg://user:pass@host/db", echo=True)

The URL format: dialect+driver://user:pass@host:port/db.

Examples:

postgresql+psycopg: Postgres via psycopg3.
postgresql+asyncpg: Postgres async.
mysql+pymysql: MySQL.
mysql+aiomysql: MySQL async.
sqlite:///app.db: SQLite.
sqlite+aiosqlite:///app.db: SQLite async.

echo=True logs SQL — useful in dev; off in prod.

Async engine

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")

Different driver; different namespace. Most of the textbook uses async because FastAPI integrates with it.

Pooling

engine = create_engine(
    URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600,
)

pool_size: persistent connections.
max_overflow: extra under load.
pool_pre_ping: check connection liveness before use (recommended).
pool_recycle: recycle connections older than N seconds (avoid stale).

For SQLite: poolclass=StaticPool (single connection in memory tests) or default for file-based.

Core: the basics

from sqlalchemy import MetaData, Table, Column, Integer, String, select, insert

metadata = MetaData()

users = Table("users", metadata,
    Column("id", Integer, primary_key=True),
    Column("email", String(255), nullable=False, unique=True),
    Column("name", String(120)),
)

with engine.connect() as conn:
    conn.execute(insert(users).values(email="[email protected]", name="Alice"))
    rows = conn.execute(select(users).where(users.c.email == "[email protected]")).all()
    conn.commit()

Table defines schema in Python.
select(), insert(), update(), delete() build statements.
conn.execute(stmt) runs them.

Core is database-agnostic at the SQL level. SQL is generated per dialect.

ORM: the basics

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)
    email: Mapped[str] = mapped_column(unique=True)
    name: Mapped[str | None]

with Session(engine) as session:
    user = User(email="[email protected]", name="Alice")
    session.add(user)
    session.commit()

with Session(engine) as session:
    result = session.execute(select(User).where(User.email == "[email protected]"))
    user = result.scalar_one()

Mapped[int] is typed; supports MyPy / Pyright. mapped_column is the column declaration with options.

Async ORM

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async with AsyncSessionLocal() as session:
    user = User(email="[email protected]")
    session.add(user)
    await session.commit()

async with AsyncSessionLocal() as session:
    result = await session.execute(select(User))
    users = result.scalars().all()

The 2.0 API is the same shape for async; just await and use AsyncSession.

Cross-DB notes

	Postgres	MySQL	SQLite
BIGSERIAL	bigserial	bigint AUTO_INCREMENT	INTEGER PRIMARY KEY
TEXT	unlimited	TEXT (limited size in some configs)	unlimited
Boolean	native	TINYINT(1)	INTEGER 0/1
JSON	native (jsonb)	json	text + parsing
ARRAY	native	not native (use JSON)	not native
Foreign keys	enforced	InnoDB enforced	optional (PRAGMA)
RETURNING	yes	partial (8.0+)	yes
ON CONFLICT	yes	INSERT … ON DUPLICATE KEY	INSERT … ON CONFLICT
Transaction isolation	full	full	serialized only

The Core dialects abstract most differences. For Postgres-specific features (JSONB, arrays, ranges, listen/notify): see the Postgres-focused textbook .

A trivial CRUD example

from sqlalchemy import select

async def get_user(session: AsyncSession, user_id: int) -> User | None:
    return await session.get(User, user_id)

async def list_users(session: AsyncSession) -> list[User]:
    result = await session.execute(select(User).order_by(User.id))
    return list(result.scalars())

async def create_user(session: AsyncSession, email: str) -> User:
    user = User(email=email)
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

async def update_user(session: AsyncSession, user_id: int, **changes) -> User:
    user = await session.get(User, user_id)
    for k, v in changes.items():
        setattr(user, k, v)
    await session.commit()
    return user

async def delete_user(session: AsyncSession, user_id: int) -> None:
    user = await session.get(User, user_id)
    await session.delete(user)
    await session.commit()

This is the mental model. Subsequent chapters dig in.

Connecting to FastAPI

from fastapi import Depends

@asynccontextmanager
async def lifespan(app):
    app.state.engine = create_async_engine(DATABASE_URL)
    app.state.sessionmaker = 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.sessionmaker() as session:
        yield session

Engine: created once at startup. Session: per-request. The textbook uses this pattern throughout.

See the FastAPI Dependency Injection chapter .

What this textbook covers

Ch	Topic
1	This chapter — Introduction + Core/ORM
2	The Engine, Pooling, and Connections
3	Schema and the Type System
4	The Session and Unit of Work
5	Querying with select()
6	Relationships, Loading Strategies
7	Transactions and Concurrency
8	Async SQLAlchemy in Depth
9	Performance: N+1, Bulk, Streaming
10	Migrations (Alembic integration)
11	Testing
12	Production patterns

For Postgres-only deep dive: companion volume.

Read this next

Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .

What SQLAlchemy is#

SQLAlchemy 2.0 changes#

Engine and dialects#

Async engine#

Pooling#

Core: the basics#

ORM: the basics#

Async ORM#

Cross-DB notes#

A trivial CRUD example#

Connecting to FastAPI#

What this textbook covers#

Read this next#