SQLAlchemy / Chapter 3 of 4

Mappings, connections, and dialects

Extend SQLAlchemy mappings and inheritance, run Core transactions, and configure dialects, isolation, and events with Python and SQL examples.

6 topicsUpdated 22 Oct 2026

18 / 27

Registry, imperative mapping, dataclasses, composites, deferred and expression attributes#

Declarative mapping is the default. Use these tools when adapting existing domain classes, tables, or value objects—or when a mapped class needs a calculated/read-only SQL attribute.

PythonImperative mapping and composite value

from dataclasses import dataclass
from sqlalchemy import Column, Integer, MetaData, Table
from sqlalchemy.orm import composite, registry

mapper_registry = registry()
account_table = Table(
    "account", mapper_registry.metadata,
    Column("id", Integer, primary_key=True), Column("name", String(50)),
)
class Account:
    def __init__(self, name: str): self.name = name
mapper_registry.map_imperatively(Account, account_table)

@dataclass
class Money:
    amount: int
    currency: str

class LineItem(Base):
    __tablename__ = "line_item"
    id: Mapped[int] = mapped_column(primary_key=True)
    amount: Mapped[int]
    currency: Mapped[str] = mapped_column(String(3))
    price: Mapped[Money] = composite(Money, "amount", "currency")

SQLSQL idea

CREATE TABLE account (id SERIAL PRIMARY KEY, name VARCHAR(50));
CREATE TABLE line_item (
  id SERIAL PRIMARY KEY, amount INTEGER NOT NULL, currency VARCHAR(3) NOT NULL
);
-- price is a Python Money(amount, currency) view over two columns.

registry()Owns mapping configuration and metadata; declarative uses it too.

map_imperatively()Map a plain class and Table separately.

MappedAsDataclassDataclass-generated constructor/repr for mapped model.

column_property()Read-only SQL expression as a model attribute.

query_expression()Per-query non-persisted expression with with_expression().

deferred()Postpone a column until access; use carefully in async.

synonym()Alternative Python name for mapped property.

Write-only collectionFor massive relationships: explicit select/insert/delete, no full collection loads.

Dynamic collectionLegacy/limited pattern that exposes query-like behavior; write-only is generally preferred.

19 / 27

Single, joined, and concrete inheritance#

Inheritance mapping represents a Python class family in relational tables. Use it when subclasses are genuinely one conceptual entity family with a stable discriminator—not simply to share a few fields.

PythonJoined-table inheritance

class Employee(Base):
    __tablename__ = "employee"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(80))
    kind: Mapped[str] = mapped_column(String(20))
    __mapper_args__ = {
        "polymorphic_on": kind, "polymorphic_identity": "employee"
    }

class Engineer(Employee):
    __tablename__ = "engineer"
    id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
    language: Mapped[str] = mapped_column(String(30))
    __mapper_args__ = {"polymorphic_identity": "engineer"}

staff = session.scalars(select(Employee)).all()

SQLEquivalent schema

CREATE TABLE employee (
  id SERIAL PRIMARY KEY, name VARCHAR(80) NOT NULL, kind VARCHAR(20) NOT NULL
);
CREATE TABLE engineer (
  id INTEGER PRIMARY KEY REFERENCES employee(id),
  language VARCHAR(30) NOT NULL
);
SELECT id, name, kind FROM employee;

Strategy	Storage	Trade-off
Single-table	One table + discriminator + nullable subtype columns.	Simple/fast; can become sparse.
Joined-table	Base table plus one table per subclass.	Normalized; subtype fields may need joins.
Concrete-table	Each subclass owns complete table.	Independent tables; polymorphic queries become unions.
Abstract/mixin	No polymorphic table family.	Share mapping/Python declarations only.

20 / 27

Hybrid properties, association proxies, ordering, mutability, versioning, and sharding#

Use hybrid properties for behavior shared by objects and queries, proxies for association access, and version columns for optimistic concurrency.

PythonHybrid works in Python and SQL

from sqlalchemy.ext.hybrid import hybrid_property

class Person(Base):
    __tablename__ = "person"
    id: Mapped[int] = mapped_column(primary_key=True)
    first_name: Mapped[str]
    last_name: Mapped[str]

    @hybrid_property
    def full_name(self):
        return f"{self.first_name} {self.last_name}"

    @full_name.expression
    def full_name(cls):
        return cls.first_name + " " + cls.last_name

people = session.scalars(
    select(Person).where(Person.full_name == "Ada Lovelace")
)

SQLEquivalent SQL

SELECT * FROM person
WHERE first_name || ' ' || last_name = 'Ada Lovelace';

PythonOptimistic concurrency version column

class Document(Base):
    __tablename__ = "document"
    id: Mapped[int] = mapped_column(primary_key=True)
    body: Mapped[str]
    version_id: Mapped[int] = mapped_column(nullable=False, default=1)
    __mapper_args__ = {"version_id_col": version_id}

# If another Session commits version 5 first, this Session's
# attempted UPDATE with expected version 4 raises StaleDataError.

SQLVersioned update SQL

UPDATE document SET body = :body, version_id = 5
WHERE id = :id AND version_id = 4;
-- Zero rows affected means stale/deleted data.

hybrid_property/methodOne behavior usable on instance and in SQL queries.

association_proxyExpose value through association object, e.g. user.role_names.

ordering_listMaintain an order-position column for a collection.

mutableTrack in-place changes inside dict/list/custom values.

indexableIndex access to structured value types.

version_id_colOptimistic locking for normal ORM flushes.

horizontal_shardAdvanced extension routes statements across database shards.

association objectLink table as a class when it has extra data.

custom collectionsInstrument a custom list/set/dict relationship collection.

Versioning does not automatically protect set-based DML.

A direct update(Document) needs its own version predicate if you require compare-and-swap behavior.

21 / 27

Core tables, connections, transactions, and direct SQL#

Core constructs SQL and executes it through database connections. Use it for reporting, ETL, bulk operations, or direct table access, with bound parameters and dialect-specific compilation.

PythonCore Table + Connection

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

metadata = MetaData()
users = Table(
    "user_account", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)), Column("email", String(255)),
)

with engine.begin() as conn:
    conn.execute(insert(users), {"name": "Ada", "email": "[email protected]"})
    result = conn.execute(
        select(users.c.id, users.c.name).where(users.c.name == "Ada")
    )
    print(result.mappings().all())

SQLEquivalent SQL

BEGIN;
INSERT INTO user_account (name, email) VALUES ('Ada', '[email protected]');
SELECT id, name FROM user_account WHERE name = 'Ada';
COMMIT;

PythonRaw SQL, savepoint, streaming

from sqlalchemy import text

with engine.connect() as conn:
    tx = conn.begin()
    try:
        conn.execute(text(
            "UPDATE product SET quantity = quantity - :n WHERE id = :id"
        ), {"n": 1, "id": 7})
        with conn.begin_nested():  # SAVEPOINT
            conn.execute(text("INSERT INTO audit_log(message) VALUES (:m)"),
                         {"m": "sold"})
        tx.commit()
    except:
        tx.rollback()
        raise

with engine.connect().execution_options(stream_results=True) as conn:
    for batch in conn.execute(select(users)).partitions(1000):
        process(batch)

SQLEquivalent SQL

BEGIN;
UPDATE product SET quantity = quantity - 1 WHERE id = 7;
SAVEPOINT sa_savepoint_1;
INSERT INTO audit_log (message) VALUES ('sold');
RELEASE SAVEPOINT sa_savepoint_1;
COMMIT;

Object	Owns	Lifetime
Engine	Dialect, driver config, connection pool.	Application-wide.
Connection	One checked-out database connection.	Short unit of direct Core work.
Transaction	Atomic commit / rollback boundary.	One logical change set.
MetaData	Table schema collection.	Definition-time/app-wide.
Table / Column	Schema and expression namespace.	Not a connection.

22 / 27

Dialects, locks, isolation, schemas, and database-specific SQL#

A dialect adapts SQLAlchemy to PostgreSQL, MySQL/MariaDB, SQLite, SQL Server, Oracle, or a third-party backend. Use portable APIs first; import from a dialect when you deliberately choose a backend feature.

PythonPostgreSQL JSONB, ARRAY, upsert

from sqlalchemy.dialects.postgresql import ARRAY, JSONB, insert

class Event(Base):
    __tablename__ = "event"
    id: Mapped[int] = mapped_column(primary_key=True)
    data: Mapped[dict] = mapped_column(JSONB)
    labels: Mapped[list[str]] = mapped_column(ARRAY(String))

by_source = select(Event).where(Event.data["source"].astext == "api")
urgent = select(Event).where(Event.labels.any("urgent"))

upsert = insert(Event).values(id=1, data={"source": "api"})
upsert = upsert.on_conflict_do_update(
    index_elements=[Event.id], set_={"data": upsert.excluded.data}
)

SQLPostgreSQL SQL

SELECT * FROM event WHERE data ->> 'source' = 'api';
SELECT * FROM event WHERE 'urgent' = ANY(labels);

INSERT INTO event (id, data) VALUES (1, '{"source":"api"}')
ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data;

PythonRow locking

with SessionLocal.begin() as session:
    product = session.scalars(
        select(Product).where(Product.id == product_id).with_for_update()
    ).one()
    if product.quantity < requested:
        raise ValueError("not enough stock")
    product.quantity -= requested

SQLEquivalent SQL

BEGIN;
SELECT * FROM product WHERE id = :product_id FOR UPDATE;
UPDATE product SET quantity = quantity - :requested WHERE id = :product_id;
COMMIT;

Isolation levelisolation_level="SERIALIZABLE" on engine/connection; behavior is backend-specific.

Schema translationschema_translate_map={None: "tenant_a"} for safe tenant schemas.

Partial/expression indexDialect index options, e.g. PostgreSQL postgresql_where.

Vendor functionfunc.vendor_function(...); check backend support.

Stored proceduretext("CALL proc(:x)") with parameters.

FOR UPDATE variantsnowait, skip_locked, of where dialect supports them.

SQLAlchemy cannot equalize database semantics.

Test exact SQL, locking, transaction isolation, JSON types, and migration behavior on the same database version used in production.

23 / 27

Events, custom compilation, custom types, and plugins#

Events fire at defined points in engine, pool, connection, session, mapper, and attribute lifecycles. Use them for small, cross-cutting infrastructure behavior—not to hide core business rules.

PythonEngine, Session, and attribute events

from sqlalchemy import event
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session

@event.listens_for(Engine, "connect")
def enable_sqlite_foreign_keys(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

@event.listens_for(Session, "before_flush")
def validate(session, flush_context, instances):
    for obj in session.new:
        if isinstance(obj, Product) and obj.price_cents < 0:
            raise ValueError("price cannot be negative")

@event.listens_for(User.name, "set", retval=True)
def normalize_name(target, value, oldvalue, initiator):
    return value.strip()

SQLSQL side effect

PRAGMA foreign_keys=ON;
-- before_flush runs before INSERT/UPDATE/DELETE reaches database.
INSERT INTO user_account (name, email) VALUES ('Ada', '[email protected]');

Event family	Examples	Typical reason
Pool/Engine	`connect`, `checkout`, `before_cursor_execute`	Connection settings and instrumentation.
Connection	`begin`, `commit`, `rollback`	Transaction observability.
Session	`before_flush`, `after_commit`, `do_orm_execute`	Validation, auditing, query policy/caching.
Mapper	`before_insert`, `after_update`	Small row lifecycle logic.
Attribute	`set`, `append`, `remove`	Normalize/validate assignment.
DDL	`before_create`, `after_drop`	Schema setup hooks.

Event safety.

Do not emit surprise ORM work from flush events unless you understand re-entrancy. Event handlers should be fast, deterministic, visible in architecture documentation, and tested.

Registry, imperative mapping, dataclasses, composites, deferred and expression attributes#

Single, joined, and concrete inheritance#

Hybrid properties, association proxies, ordering, mutability, versioning, and sharding#

Core tables, connections, transactions, and direct SQL#

Dialects, locks, isolation, schemas, and database-specific SQL#

Events, custom compilation, custom types, and plugins#

Help improve CodeRef