SQLAlchemy, Transactions & Migrations

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production.

The reference material consulted while preparing these pages is listed in the section bibliography.

This page is written against Pyramid 2.1 on Python 3.13, checked on 2026-10-07, with SQLAlchemy 2.0.54 (the 2.0 maintenance series) and SQLAlchemy 2.1.3 (released 2026-09-24 as 2.1.0, the current series), pyramid_tm 2.6, pyramid_retry 2.1.1, zope.sqlalchemy 4.1, transaction 5.1 and Alembic 1.20.0. Every Python block that is not marked otherwise was executed with webtest.TestApp against SQLite, on both SQLAlchemy series, and the commands and migration scripts were produced by a project generated from the 2.1-branch of the cookiecutter with backend=sqlalchemy. Pyramid itself contains no database code: the data layer is a small stack of independent packages that the cookiecutter wires together, and this page explains each layer, the order in which they act on a request, and how the schema is evolved with migrations. The scenario is Bookshelf Shelves: shelves with a slug and a title, each holding books. The Flask counterparts are Flask-SQLAlchemy and Database Migrations with Flask-Migrate; the Pyramid answer to "who commits?" is different, and that difference is the main subject here.

The Pieces and Who Does What

A Pyramid application that talks to a relational database through SQLAlchemy is built from six packages. Nothing in the list is Pyramid-specific except pyramid_tm and pyramid_retry, which is why the same stack also works for ZODB and for any other resource that can join a transaction.

Package Role Where it shows up

SQLAlchemy

The ORM and the SQL toolkit: declarative models, the Session unit of work, engines and connection pools.

models/, every view that touches data.

transaction

The generic transaction manager from the Zope world: a transaction can have several data managers joined to it and they all commit or all abort.

request.tm.

zope.sqlalchemy

A data manager that joins a SQLAlchemy Session to a transaction transaction, so that committing the transaction commits the session.

get_tm_session() in models/__init__.py.

pyramid_tm

A tween that begins a transaction when a request enters the application and commits or aborts it when the response leaves.

config.include('pyramid_tm'), request.tm, the tm.* settings.

pyramid_retry

An execution policy that runs the whole request again, with a fresh request object, when it failed with a retryable error.

config.include('pyramid_retry'), retry.attempts.

Alembic

Schema migrations for SQLAlchemy metadata: revision scripts, upgrade and downgrade, and autogeneration.

The alembic/ folder, the [alembic] section of the .ini files, the alembic command.

Source: Pyramid wiki2 tutorial: Basic Layout

Three design decisions follow from this stack and shape the rest of the page.

  • Views never call commit(). The transaction belongs to the request, not to the view. A view adds, changes and deletes objects; the tween commits when the view returns normally and rolls back when it raises. This is the opposite of the Flask-SQLAlchemy habit of an explicit db.session.commit(), and calling dbsession.commit() by hand is actually refused by zope.sqlalchemy (see Pitfalls and Security).

  • The session is per request. request.dbsession creates a new Session for the request the first time it is read and the transaction manager closes it at the end. There is no global or thread-local session.

  • Retries restart everything. When a request is retried, the router builds a new request, request.dbsession creates a new session, and a new transaction begins, so no state from the failed attempt survives.

The next figure shows the order of events for one request. The retry policy is the outermost layer, pyramid_tm sits over the exception-view (excview) tween, and the session is created lazily by the view, not by the tween.

sequenceDiagram participant C as Client participant R as retry policy participant T as pyramid_tm tween participant X as excview tween participant V as View participant S as Session C->>R: HTTP request R->>T: attempt n of retry.attempts T->>T: request.tm.begin() T->>X: handler(request) X->>V: view lookup and call V->>S: request.dbsession creates the session and joins the transaction S-->>V: rows, flush on demand V-->>X: response, or an exception for an exception view X-->>T: response alt doomed, exception seen, or commit veto T->>S: request.tm.abort() rolls back and closes the session else clean response T->>S: request.tm.commit() flushes, COMMIT, closes the session end T-->>R: response R-->>C: response, or a new attempt when the error was retryable
Figure 1. The life of a transaction around one request

The tween makes the decision between commit and abort after the response exists, which is why a response can be inspected before it is allowed to commit (a commit veto). The decision itself is a short chain of checks, taken from the pyramid_tm 2.6 source and its documentation:

flowchart TD A["The downstream handler returned"] --> B{"Doomed with request.tm.doom()?"} B -- yes --> AB["abort, return the response"] B -- no --> C{"tm.commit_veto set and returns True?"} C -- yes --> AB C -- no --> D{"An exception view handled an exception, and no commit veto is configured?"} D -- yes --> AB D -- no --> CM["commit"] E["The downstream handler raised"] --> F["abort, re-raise"]
Figure 2. How pyramid_tm chooses between commit and abort

Source: pyramid_tm documentation, transaction usage

The Cookiecutter sqlalchemy Backend

The cookiecutter generates everything above when it is asked for backend=sqlalchemy; the full question list and the other backends are on Cookiecutter Projects & pserve. With the project name Bookshelf Shelves and the Jinja2 template language:

cookiecutter gh:Pylons/pyramid-cookiecutter-starter --checkout 2.1-branch
# project_name: Bookshelf Shelves, template_language: jinja2, backend: sqlalchemy
cd bookshelf_shelves
python3 -m venv env
env/bin/pip install -e ".[testing]"
env/bin/alembic -c development.ini revision --autogenerate -m "init"
env/bin/alembic -c development.ini upgrade head
env/bin/initialize_bookshelf_shelves_db development.ini
env/bin/pytest
env/bin/pserve development.ini

Source: pyramid-cookiecutter-starter (the generated README.md lists these steps)

The first two Alembic commands create the first revision from the models and apply it; the project does not ship a revision, because the model it ships (MyModel) is a placeholder that you are expected to replace. The files that matter, as generated:

File Content

bookshelf_shelves/models/meta.py

Base, a DeclarativeBase subclass with a metadata naming_convention that Alembic autogeneration relies on.

bookshelf_shelves/models/mymodel.py

The sample model, to be replaced by your own modules (this page uses shelf.py).

bookshelf_shelves/models/__init__.py

get_engine, get_session_factory, get_tm_session and includeme, the wiring explained in Wiring: The Engine, the Session and request.dbsession. It also imports every model module, so that all tables are attached to Base.metadata, and calls configure_mappers().

bookshelf_shelves/alembic/

env.py, script.py.mako and an empty versions/ folder.

bookshelf_shelves/scripts/initialize_db.py

The console script initialize_bookshelf_shelves_db, declared under [project.scripts] in pyproject.toml.

bookshelf_shelves/pshell.py

The setup hook named in the [pshell] section: it begins a transaction and injects tm, dbsession and models into the pshell namespace.

development.ini, production.ini, testing.ini

sqlalchemy.url, retry.attempts, the [alembic] section and the sqlalchemy and alembic loggers.

tests/conftest.py

The fixtures dbengine, app, tm, dbsession and testapp (see Testing with Transaction Rollback).

Source: pyramid-cookiecutter-starter, 2.1-branch

The official SQLAlchemy and URL dispatch wiki tutorial ("wiki2") is generated from exactly this project and is the best companion to this page: its Basic Layout, Defining the Domain Model and Testing chapters explain the same files with a User and Page model, and it adds authentication and authorization on top. Read it after this page, or alongside, and note that its commands use $VENV/bin/..., which is the same as the env/bin/... above. Source: Pyramid: SQLAlchemy + URL Dispatch Wiki Tutorial

Models with SQLAlchemy 2.0

The generated models use the SQLAlchemy 2.0 declarative style: a DeclarativeBase subclass as the base, annotated attributes with Mapped[...], and mapped_column() for the options that a type annotation cannot express. A column is NOT NULL when its annotation is not Optional (or X | None), so the annotation is the nullability. The block below defines the metadata and the two models of Bookshelf Shelves. The __tablename__ attribute names the table; the relationship() pair connects a shelf to its books, with cascade="all, delete-orphan" so that deleting a shelf deletes its books.

from sqlalchemy import ForeignKey, MetaData, true
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    # The naming convention gives every constraint a predictable name, which
    # Alembic needs to drop or alter constraints later on.
    metadata = MetaData(naming_convention={
        "ix": "ix_%(column_0_label)s",
        "uq": "uq_%(table_name)s_%(column_0_name)s",
        "ck": "ck_%(table_name)s_%(constraint_name)s",
        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
        "pk": "pk_%(table_name)s",
    })


class Shelf(Base):
    __tablename__ = "shelves"
    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column(unique=True)
    title: Mapped[str]
    is_public: Mapped[bool] = mapped_column(server_default=true())
    books: Mapped[list["Book"]] = relationship(
        back_populates="shelf", cascade="all, delete-orphan"
    )


class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    shelf_id: Mapped[int] = mapped_column(ForeignKey("shelves.id"))
    title: Mapped[str]
    isbn: Mapped[str | None]
    shelf: Mapped[Shelf] = relationship(back_populates="books")


print(sorted(c.name for c in Shelf.__table__.constraints))
assert Book.__table__.c.isbn.nullable and not Book.__table__.c.title.nullable

Source: SQLAlchemy 2.0: Declarative Table Configuration with Mapped and mapped_column

The output is ['pk_shelves', 'uq_shelves_slug']. Three details are worth keeping in mind.

  • The naming convention is not optional in practice. Without it, SQLite and PostgreSQL name anonymous constraints differently, and an Alembic migration cannot refer to a constraint it has no name for. The convention in the block is the one from the Alembic documentation; the generated meta.py uses the same keys and differs only in its ck template, which wraps the constraint name in backticks.

  • Every model module must be imported. The metadata only knows the tables of the modules that have been imported. That is the reason for from .mymodel import MyModel in the generated models/__init__.py, and for the configure_mappers() call after it: it resolves all relationship strings such as "Book" once, at start-up, so a typo fails when the application is created and not on the first request.

  • server_default is a database default. is_public above carries DEFAULT 1 in the CREATE or ALTER statement, which matters for migrations of tables that already contain rows (see What Autogenerate Cannot Do).

Older tutorials and the first Pyramid scaffolds use the 1.x style: declarative_base(), Column(Integer, ...) attributes and DBSession.query(Model). It still works, but the 2.0 style is what the 2.1-branch cookiecutter and the current wiki2 tutorial generate, it is typed, and it is what this page uses. The query API changes with it: the Query object of session.query() is legacy, and select() with session.scalars() or session.execute() is the current form.

SQLAlchemy 2.1 requires Python 3.11 or newer, and Pyramid 2.1 still supports Python 3.10, so on Python 3.10 pip installs the 2.0 series. The models above are unchanged between the two series. Two 2.1 changes can still affect a Pyramid project: session autoflush behaviour was simplified to be unconditional, and a plain postgresql:// URL now selects the psycopg 3 driver instead of psycopg2. Write postgresql+psycopg2:// in sqlalchemy.url if the project must keep using psycopg2 after an upgrade. Source: SQLAlchemy 2.1: What’s New in SQLAlchemy 2.1

Wiring: The Engine, the Session and request.dbsession

The generated models/__init__.py has four functions. The block below is a condensed, single-file copy that runs as written (the project version adds a long docstring to get_tm_session).

import transaction
import zope.sqlalchemy
from sqlalchemy import engine_from_config
from sqlalchemy.orm import sessionmaker


def get_engine(settings, prefix="sqlalchemy."):
    return engine_from_config(settings, prefix)


def get_session_factory(engine):
    factory = sessionmaker()
    factory.configure(bind=engine)
    return factory


def get_tm_session(session_factory, transaction_manager, request=None):
    dbsession = session_factory(info={"request": request})
    zope.sqlalchemy.register(dbsession, transaction_manager=transaction_manager)
    return dbsession


def includeme(config):
    settings = config.get_settings()
    settings["tm.manager_hook"] = "pyramid_tm.explicit_manager"
    config.include("pyramid_tm")
    config.include("pyramid_retry")

    # hook to share one engine, for example the one of a test fixture
    dbengine = settings.get("dbengine")
    if not dbengine:
        dbengine = get_engine(settings)

    session_factory = get_session_factory(dbengine)
    config.registry["dbsession_factory"] = session_factory

    def dbsession(request):
        # hook to share a session in tests (see the testing section)
        dbsession = request.environ.get("app.dbsession")
        if dbsession is None:
            dbsession = get_tm_session(session_factory, request.tm, request=request)
        return dbsession

    config.add_request_method(dbsession, reify=True)

Source: Pyramid wiki2 tutorial: Basic Layout, Content Models with models and Database Configuration

Reading it from the top:

  • get_engine turns the sqlalchemy.* keys of the .ini file into an Engine with engine_from_config. The engine owns the connection pool and is created once, at start-up.

  • get_session_factory is a plain sessionmaker, not a scoped_session. A thread-local session registry would add nothing, because each request gets its own session from the factory.

  • get_tm_session creates a session and calls zope.sqlalchemy.register(session, transaction_manager=...), which attaches the SQLAlchemy event listeners that join the session to the transaction when it begins and mark it as changed when it flushes. The info={"request": request} entry makes the active request reachable from any mapped object through the session, which is occasionally useful in event listeners.

  • includeme is the standard config.include hook described on Configuration & the Configurator. It sets tm.manager_hook to the dotted name pyramid_tm.explicit_manager before including pyramid_tm, includes pyramid_retry, builds the session factory, stores it in the registry, and adds the reified request property request.dbsession. A reified property is computed on first access and cached on the request (see Request & Response), so a request that never touches the database never opens a connection.

The tm.manager_hook line is the important one. By default request.tm is the thread-local transaction.manager; with explicit_manager each request gets its own TransactionManager(explicit=True). In explicit mode a transaction exists only between begin() and commit() or abort(), and code that touches the manager outside that window raises NoTransaction instead of silently opening an implicit transaction that nobody commits. The pyramid_tm documentation recommends this mode for that reason. Source: pyramid_tm documentation, Custom Transaction Managers

Engine and Pool Settings

Every key of the .ini file that starts with sqlalchemy. is passed to create_engine after the prefix is removed, so sqlalchemy.url, sqlalchemy.pool_size and the other documented engine arguments all work. The values come from the file as strings; engine_from_config converts only a short list of them.

Setting Meaning

sqlalchemy.url

The database URL. The cookiecutter writes sqlite:///%(here)s/bookshelf_shelves.sqlite; PasteDeploy expands %(here)s to the directory of the .ini file.

sqlalchemy.echo

Log every SQL statement (true, false or debug). Converted. Never leave it on in production, because the statements and their parameters reach the logs.

sqlalchemy.pool_size, sqlalchemy.max_overflow

Size of the connection pool and how many extra connections it may open under load (defaults 5 and 10 for the default QueuePool). Converted to integers.

sqlalchemy.pool_timeout

Seconds to wait for a free connection before failing. Converted.

sqlalchemy.pool_recycle

Replace connections older than this many seconds, for servers that close idle connections. Converted.

sqlalchemy.pool_pre_ping

Test a connection when it is checked out and replace it if it died. Not converted, see the pitfall below.

Source: SQLAlchemy 2.0: engine_from_config

The list of converted keys is the one in the dialect (pool_timeout, echo, echo_pool, pool_recycle, pool_size, max_overflow and future), and the next block shows what that means for a boolean that is not on it.

from sqlalchemy import engine_from_config

engine = engine_from_config(
    {
        "sqlalchemy.url": "sqlite://",
        "sqlalchemy.echo": "false",
        "sqlalchemy.pool_pre_ping": "false",
    },
    "sqlalchemy.",
)
print(repr(engine.echo), repr(engine.pool._pre_ping))
assert engine.echo is False
assert engine.pool._pre_ping == "false"   # still the string: truthy, so pre-ping is ON

Source: SQLAlchemy 2.0: engine_from_config

So sqlalchemy.pool_pre_ping = false in an .ini file enables pre-ping, because the non-empty string 'false' is truthy. To turn a non-converted option off, leave the key out; to set it conditionally, build the engine in code with get_engine(settings, pool_pre_ping=True)-style wrappers of your own. The same holds for any boolean option that is not in the converted list.

A second sizing rule comes from the WSGI server. Waitress runs 4 worker threads by default, and each in-flight request holds at most one connection from the pool, so the default pool of 5 plus 10 overflow connections is enough for one Waitress process; with Gunicorn workers each process has its own pool, and the total of all pools must stay below the connection limit of the database server. Source: SQLAlchemy 2.0: Connection Pooling

The Database URL Is a Secret

A database URL of the form scheme, user, password, host and database name contains a password and must not live in a file that is committed. The generated .ini files contain only the SQLite file path, which is harmless, but a production file should not hold a password either. The simplest pattern is a main function that lets the environment override the setting, so the file keeps a default for development and the real URL is injected by the deployment (see Settings, Environment & .ini Files):

import os


def with_env_url(settings):
    """Let DATABASE_URL override sqlalchemy.url; used at the top of main()."""
    settings = dict(settings)
    settings["sqlalchemy.url"] = os.environ.get("DATABASE_URL", settings["sqlalchemy.url"])
    return settings


os.environ["DATABASE_URL"] = "sqlite://"
print(with_env_url({"sqlalchemy.url": "sqlite:///dev.sqlite"})["sqlalchemy.url"])
del os.environ["DATABASE_URL"]
assert with_env_url({"sqlalchemy.url": "sqlite:///dev.sqlite"})["sqlalchemy.url"] == "sqlite:///dev.sqlite"

Source: Pyramid: Environment Variables and .ini File Settings

The Alembic env.py of the cookiecutter reads the same .ini file, so a migration run needs the same override: edit its settings = get_appsettings(...) line to apply with_env_url as well, or run the command with a copy of the file that points at the right database. Use a database account with the least privileges that fit the job: the running application needs SELECT, INSERT, UPDATE and DELETE, while the migration account also needs the right to alter the schema, and the two should not be the same account in production.

Using the Session in Views

The block below assembles a complete application from the pieces above (it reuses the models and the includeme of the previous blocks): an in-memory SQLite database, the includeme that installs pyramid_tm, pyramid_retry and request.dbsession, and three views. Note what the views do not contain: no commit(), no rollback(), no close().

from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPNotFound
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session, selectinload
from webtest import TestApp


def list_shelves(request):
    query = select(Shelf).options(selectinload(Shelf.books)).order_by(Shelf.slug)
    return [
        {"slug": s.slug, "title": s.title, "books": len(s.books)}
        for s in request.dbsession.scalars(query)
    ]


def add_shelf(request):
    data = request.json_body
    shelf = Shelf(slug=data["slug"], title=data["title"])
    request.dbsession.add(shelf)
    request.dbsession.flush()       # INSERT now: assigns shelf.id, surfaces constraint errors
    request.response.status = 201
    return {"id": shelf.id, "slug": shelf.slug}


def add_book(request):
    slug = request.matchdict["slug"]
    shelf = request.dbsession.scalar(select(Shelf).where(Shelf.slug == slug))
    if shelf is None:
        raise HTTPNotFound(f"no shelf {slug}")
    book = Book(title=request.json_body["title"], shelf=shelf)
    request.dbsession.add(book)
    request.dbsession.flush()
    request.response.status = 201
    return {"id": book.id, "shelf": shelf.slug}


def count(engine, model=Shelf):
    """Rows seen by a separate session: the committed state of the database."""
    with Session(engine) as session:
        return session.scalar(select(func.count()).select_from(model))


def make_app(extra_routes=(), url="sqlite://", **extra_settings):
    engine = create_engine(url)
    Base.metadata.create_all(engine)
    settings = {"dbengine": engine, **extra_settings}
    with Configurator(settings=settings) as config:
        config.include(includeme)
        config.add_route("shelves", "/shelves")
        config.add_route("books", "/shelves/{slug}/books")
        config.add_view(list_shelves, route_name="shelves", request_method="GET", renderer="json")
        config.add_view(add_shelf, route_name="shelves", request_method="POST", renderer="json")
        config.add_view(add_book, route_name="books", request_method="POST", renderer="json")
        for configure in extra_routes:
            configure(config)
        app = config.make_wsgi_app()
    return app, engine


app, engine = make_app()
client = TestApp(app)
client.post_json("/shelves", {"slug": "ada", "title": "Ada's shelf"}, status=201)
client.post_json("/shelves/ada/books", {"title": "Dune"}, status=201)
client.post_json("/shelves/nobody/books", {"title": "Ghost"}, status=404)
print(client.get("/shelves").json)
assert client.get("/shelves").json == [{"slug": "ada", "title": "Ada's shelf", "books": 1}]
assert count(engine) == 1 and count(engine, Book) == 1

Source: Pyramid wiki2 tutorial: Defining Views

Points that the block demonstrates, and the corresponding rules:

  • flush() is not commit(). flush sends the pending INSERT so that the primary key is known and constraint violations surface in the view, where they can be handled. The transaction is still open and the data is invisible to other connections until the tween commits. count(engine) uses a separate session to prove what was committed.

  • selectinload avoids the N+1 pattern. Without it, len(s.books) would issue one SELECT per shelf; with it, all the books of the page are loaded in one extra query. Lazy loading still works, but it is the default cause of slow list views.

  • A raised HTTP exception rolls back. The 404 view above raised HTTPNotFound before it wrote anything, so there was nothing to roll back, but had it written first, the transaction would be aborted all the same: an exception view produces the response and pyramid_tm sees that the request carried an exception (see the second figure and the section on exception views below).

  • Objects must not outlive the transaction. After the commit the session is closed, so ORM objects are detached and loading an unloaded attribute raises DetachedInstanceError. The json renderer runs inside the view tween, before the commit, which is why returning dictionaries built from objects is safe; returning the objects and serializing them later (in a response callback or a middleware) is not. Convert to plain data in the view.

  • Never build a model from the whole request body. Shelf(**request.json_body) would let a client set any column, such as is_public or an id. Pick the allowed fields by name, as add_shelf does, or validate with a schema (see Renderers & JSON for the JSON side).

One Transaction per Request: Commit, Abort, Doom and Veto

pyramid_tm begins a transaction with request.tm.begin() when the request enters, and decides at the end. The rules are those of the second figure; the table gives the vocabulary and the knob for each.

Situation Result How to change it

The view returns a response, nothing else happened

request.tm.commit()

This is the default.

The view (or something under the tween) raised an exception that no exception view handled

request.tm.abort(), then the exception propagates

Fix the cause, or add an exception view.

An exception view turned an exception into a response

abort() (the response is returned), unless a commit veto is configured

tm.commit_veto.

request.tm.doom() was called

abort(), with the response kept

Do not call doom(); see below.

tm.commit_veto is set and returns true for (request, response)

abort()

pyramid_tm.default_commit_veto or your own dotted name.

The view returned (not raised) an HTTPBadRequest, HTTPForbidden or a 500 response

Commit, unless a veto says otherwise

tm.commit_veto = pyramid_tm.default_commit_veto.

Source: pyramid_tm documentation, transaction usage and commit veto

Doom marks a transaction as unable to commit: any later commit() raises, and the tween aborts it at the end of the request while still returning the response. It is the tool for a view that wants to do work, look at the result and discard it, such as a "preview" endpoint. The sharpest edge of the table is the last row: returning an HTTP error object is just returning a response, so the work commits. The next block shows doom, the default behaviour of a returned error, and the veto that changes it. It uses the application of the previous block with three extra routes.

from pyramid.httpexceptions import HTTPBadRequest


def preview(request):
    request.dbsession.add(Shelf(slug="scratch", title="Not kept"))
    request.dbsession.flush()
    request.tm.doom()                 # keep the response, discard the work
    return {"previewed": True}


def returned_error(request):
    request.dbsession.add(Shelf(slug="half-done", title="Oops"))
    request.dbsession.flush()
    return HTTPBadRequest("rejected after writing")   # returned, not raised


def broken(request):
    request.dbsession.add(Shelf(slug="lost", title="Rolled back"))
    request.dbsession.flush()
    raise RuntimeError("bug after the write")


def tx_routes(config):
    for name, view in (("preview", preview), ("returned", returned_error), ("broken", broken)):
        config.add_route(name, f"/{name}")
        config.add_view(view, route_name=name, renderer="json")


# 1. doom: response returned, nothing committed
app, engine = make_app([tx_routes])
client = TestApp(app)
assert client.get("/preview").json == {"previewed": True}
assert count(engine) == 0

# 2. an exception rolls back and propagates
try:
    client.get("/broken")
except RuntimeError as exc:
    print("raised:", exc)
assert count(engine) == 0

# 3. a returned HTTP error commits ...
client.get("/returned", status=400)
assert count(engine) == 1

# 4. ... unless a commit veto is configured
app, engine = make_app([tx_routes], **{"tm.commit_veto": "pyramid_tm.default_commit_veto"})
TestApp(app).get("/returned", status=400)
assert count(engine) == 0
print("doom, exception and veto all discarded the write")

Source: pyramid_tm documentation, transaction usage and commit veto

The default veto aborts on any status that starts with 4 or 5, and on a response header X-Tm whose value is not commit. The header lets a view override the veto in both directions: X-Tm: abort forces a rollback of a 200, and X-Tm: commit allows a commit of an error response, such as a 409 that has recorded the conflict. A custom veto is a function (request, response) returning true to abort. Since pyramid_tm 2.2 the veto also decides the fate of a handled exception: a 3xx or 2xx response rendered from a raised exception (for example a raised HTTPFound) commits under the default veto, whereas 4xx and 5xx responses abort. Source: pyramid_tm changes, 2.2

The tm.* Settings and the Environ Keys

Setting Meaning

tm.manager_hook

Dotted name (or callable, when set in code) of a function (request) that returns the transaction manager. The cookiecutter sets pyramid_tm.explicit_manager. Default: the thread-local transaction.manager.

tm.commit_veto

Dotted name of (request, response) returning true to abort. Default: none. The older name pyramid_tm.commit_veto is still read as a fallback.

tm.activate_hook

Dotted name of (request) returning false to leave a request unmanaged, for example a long-poll endpoint that must not hold a transaction open.

tm.annotate_user

When true (default), the tween reads request.authenticated_userid and stores it on the transaction as its user. Set false to avoid touching the security policy (and, with it, the database) at the start of every request.

tm.attempts

Removed in pyramid_tm 2.0; setting it only produces a warning. Use retry.attempts of pyramid_retry.

Source: pyramid_tm documentation, Adding an Activation Hook, Adding a Commit Veto Hook, Avoid Accessing the Authentication Policy

Two keys of the WSGI environ complete the picture, and they are how the test fixtures work: if tm.active is already set, pyramid_tm does nothing for the request, and tm.manager, when present, replaces request.tm. Both are documented as the way to take over transaction handling in tests, see Testing with Transaction Rollback. If the repoze.tm or repoze.tm2 middleware is in the WSGI pipeline, pyramid_tm also steps aside.

Exception Views and tm_active

Because the tween is positioned over the excview tween, the transaction is normally still open while an exception view runs, but that is not guaranteed: an error raised by a tween that is itself over pyramid_tm, or while committing, is rendered after the transaction ended. An exception view that needs the transaction (for example to look up who the authenticated user is) must say so with the tm_active=True view predicate and have a fallback without it. The block below registers both for one exception type and shows which one runs. It also shows the IntegrityError pattern: a unique-constraint violation found at flush() is turned into a 409 response. The session is unusable after a failed flush, so that exception view must not use request.dbsession.

from sqlalchemy.exc import IntegrityError


def duplicate_slug(exc, request):
    request.response.status = 409
    return {"error": "slug already exists"}       # no database access here


def tm_open(exc, request):
    request.response.status = 500
    return {"error": "boom", "transaction": "open"}


def tm_closed(exc, request):
    request.response.status = 500
    return {"error": "boom", "transaction": "closed"}


def exc_routes(config):
    config.add_exception_view(duplicate_slug, context=IntegrityError, renderer="json")
    config.add_exception_view(tm_open, context=RuntimeError, tm_active=True, renderer="json")
    config.add_exception_view(tm_closed, context=RuntimeError, renderer="json")
    config.add_route("broken", "/broken")
    config.add_view(broken, route_name="broken", renderer="json")


app, engine = make_app([exc_routes])
client = TestApp(app)
client.post_json("/shelves", {"slug": "ada", "title": "A"}, status=201)
dup = client.post_json("/shelves", {"slug": "ada", "title": "B"}, status=409)
print(dup.json)
assert count(engine) == 1                          # the first shelf survived, the second never committed
body = client.get("/broken", status=500).json
print(body)
assert body["transaction"] == "open" and count(engine) == 1

Source: pyramid_tm documentation, Error Handling and View Predicates

Work that must survive a failure, such as writing an audit or error record when a request fails, cannot use request.dbsession: it joins the transaction that is about to be aborted. Use a separate connection or session, outside the transaction manager. The general rules for exception views, tm_active and pyramid_retry are on Errors & Exception Views.

Savepoints and Partial Failure

A view sometimes wants to try something that may fail and carry on: insert a row, and if it already exists, continue without it. In a transaction, a failed statement poisons the rest, so the answer is a savepoint. pyramid_tm says to take it from the transaction manager, sp = request.tm.savepoint(), and to roll back with sp.rollback(), and not to call session.begin_nested() directly: the manager synchronizes the savepoint across all data managers joined to the transaction, and zope.sqlalchemy (since version 3.0) no longer allows session.commit() inside a savepoint. The shape of the code, for a database with savepoints such as PostgreSQL, is:

# norun  (needs a database with SAVEPOINT support, such as PostgreSQL)
from sqlalchemy.exc import IntegrityError


def add_shelf_if_missing(request, slug, title):
    sp = request.tm.savepoint()
    try:
        request.dbsession.add(Shelf(slug=slug, title=title))
        request.dbsession.flush()
    except IntegrityError:
        sp.rollback()              # only the failed INSERT is undone
        return False
    return True

Source: pyramid_tm documentation, Savepoints

zope.sqlalchemy keeps a list of drivers without savepoint support, and the only one on it is sqlite. On such a database request.tm.savepoint() raises TypeError("Savepoints unsupported", ...), and the failure is worse than an exception: the transaction package records it and marks the transaction as failed, so catching the TypeError and carrying on does not help, because the later commit raises TransactionFailedError and a doomed transaction cannot be recovered either. The next block shows the error in a project that uses SQLite, which is the default of the cookiecutter:

def try_savepoint(request):
    request.dbsession.scalar(select(1))      # the session joins the transaction on first use
    request.tm.savepoint()                   # TypeError on SQLite
    return {"unreachable": True}


def sp_routes(config):
    config.add_route("sp", "/sp")
    config.add_view(try_savepoint, route_name="sp", renderer="json")


app, engine = make_app([sp_routes])
try:
    TestApp(app).get("/sp")
except TypeError as exc:
    print(exc.args[0])
    assert exc.args[0] == "Savepoints unsupported"
else:
    raise AssertionError("SQLite has no savepoint support in zope.sqlalchemy")

Source: zope.sqlalchemy on PyPI

The consequence is practical: code that uses request.tm.savepoint() cannot be exercised against SQLite, so the test database of a project that needs savepoints must be the same kind of server as the production one. Not every data manager supports savepoints, and changes joined to a transaction by one that does not cannot be rolled back individually.

Raw SQL, mark_changed and the Session State

zope.sqlalchemy avoids useless commits: a session that was only read from is rolled back at the end, not committed. It learns that data changed from the session events (after_flush, bulk update and delete, and, since SQLAlchemy 1.4, ORM-enabled INSERT, UPDATE and DELETE statements). A statement that SQLAlchemy cannot classify, such as session.execute(text("UPDATE ...")), is not seen as a write, because there is no way to guess whether a string of SQL changes data. The session then counts as unchanged and is rolled back, and the update silently disappears. The fix is to call mark_changed(session) after the statement, or to register the session with initial_state="changed".

from sqlalchemy import text
from zope.sqlalchemy import mark_changed


def retitle_lost(request):
    request.dbsession.execute(text("UPDATE shelves SET title = 'Renamed' WHERE slug = 'ada'"))
    return {"ok": True}


def retitle_kept(request):
    request.dbsession.execute(text("UPDATE shelves SET title = 'Renamed' WHERE slug = 'ada'"))
    mark_changed(request.dbsession)
    return {"ok": True}


def raw_routes(config):
    for name, view in (("lost", retitle_lost), ("kept", retitle_kept)):
        config.add_route(name, f"/{name}")
        config.add_view(view, route_name=name, renderer="json")


def title_of(engine):
    with Session(engine) as session:
        return session.scalar(select(Shelf.title).where(Shelf.slug == "ada"))


app, engine = make_app([raw_routes])
client = TestApp(app)
client.post_json("/shelves", {"slug": "ada", "title": "Ada's shelf"}, status=201)
client.get("/lost")
print("without mark_changed:", title_of(engine))
assert title_of(engine) == "Ada's shelf"
client.get("/kept")
print("with mark_changed:", title_of(engine))
assert title_of(engine) == "Renamed"

Source: zope.sqlalchemy on PyPI, Full Example

Other zope.sqlalchemy behaviours to know, all from its README and source:

  • The session is closed after the commit unless register(..., keep_session=True) is used, and the objects it loaded are then detached. keep_session=True is meant for test suites and long-lasting sessions; the session must then be closed by hand.

  • session.commit() is refused. The before_commit listener asserts that the transaction is committed through the transaction manager, with the message "Transaction must be committed using the transaction manager". A view that calls request.dbsession.commit() fails at that line, which is the intended guard.

  • Two-phase commit. When the session is created with twophase=True, zope.sqlalchemy uses a two-phase data manager, so that several databases (and other resource managers) can commit atomically. Few applications need it.

  • Connections obtained from the engine directly (not through the session) are outside the transaction integration and must be committed by their own code. Source: zope.sqlalchemy on PyPI

Transient Errors: pyramid_retry

Some failures say nothing about the request and are cured by trying again: a deadlock, a serialization failure under SERIALIZABLE isolation, an optimistic-concurrency conflict. pyramid_retry is an execution policy (the object that calls the router for each request) that runs the whole request again, up to retry.attempts times, when it ends with a retryable error. Including it is one line, and the default number of attempts is 3:

[app:main]
use = egg:bookshelf_shelves
retry.attempts = 3

Source: pyramid_retry documentation, Usage

An error is retryable when it is a pyramid_retry.RetryableException, or when it provides the marker interface pyramid_retry.IRetryableError. Beyond that, the packages of this page mark errors on their own:

  • pyramid_tm marks every transaction.interfaces.TransientError as retryable, and tags the exception that the datamanagers report as retryable before it aborts the transaction (the information is gone afterwards).

  • zope.sqlalchemy reports as retryable a SQLAlchemy ConcurrentModificationError, and DBAPIError`s whose original exception is a psycopg2 `TransactionRollbackError, a psycopg 3 error with a SQLSTATE that starts with 40 (this includes deadlocks and serialization failures), an Oracle ORA-08177, or MySQL error 1213 through PyMySQL.

  • Your own errors: pyramid_retry.mark_error_retryable(SomeError) marks a class or an instance. Mark the narrowest type that is genuinely transient; marking SQLAlchemy’s OperationalError as a whole would retry connection refusals and syntax errors as well.

SQLite does not produce any of these errors, so the example below raises a transaction.interfaces.TransientError by hand to show the mechanics: the first attempts fail, each attempt starts from a clean slate (the rows written by the failed attempts are gone), and the third attempt succeeds. A second route always fails, and an exception view with the last_retry_attempt=True predicate turns the last failure into a 503 instead of an error page.

from transaction.interfaces import TransientError

attempts = []


def flaky(request):
    attempts.append(len(attempts) + 1)
    request.dbsession.add(Shelf(slug=f"try-{attempts[-1]}", title="x"))
    request.dbsession.flush()
    if len(attempts) < 3:
        raise TransientError("serialization failure, try again")
    return {"attempt": attempts[-1], "rows_in_this_transaction": count_in_session(request)}


def count_in_session(request):
    return request.dbsession.scalar(select(func.count()).select_from(Shelf))


def always_failing(request):
    raise TransientError("the database is busy")


def busy(exc, request):
    request.response.status = 503
    request.response.headers["Retry-After"] = "5"
    return {"error": "busy"}


def retry_routes(config):
    config.add_route("flaky", "/flaky")
    config.add_route("busy", "/busy")
    config.add_view(flaky, route_name="flaky", renderer="json")
    config.add_view(always_failing, route_name="busy", renderer="json")
    # runs only on the last attempt: earlier attempts are retried instead
    config.add_exception_view(busy, context=TransientError, last_retry_attempt=True, renderer="json")


app, engine = make_app([retry_routes], **{"retry.attempts": "3"})
client = TestApp(app)
body = client.get("/flaky").json
print(body, "attempts:", attempts)
assert attempts == [1, 2, 3] and body["rows_in_this_transaction"] == 1
assert count(engine) == 1                 # only the successful attempt committed
assert client.get("/busy", status=503).headers["Retry-After"] == "5"

Source: pyramid_retry documentation, Usage, Custom Retryable Errors and View Predicates

Retrying has costs and rules that the documentation states and that decide whether a view is safe to run again:

  • The view must be idempotent apart from the transaction. Everything inside the transaction is rolled back, but a call to an external service, an email or a counter in memory is not. Do such side effects in a NewResponse subscriber, a response callback, or a transaction hook, so that they only run for the attempt that committed. A transaction-aware mail sender is the usual example.

  • The request body is copied. To make a retry possible, request.make_body_seekable() copies the body of the request, which can be a large copy for uploads. Set the per-request number of attempts to 1 with the retry.activate_hook setting (a function (request) returning a number of attempts or None) for such endpoints.

  • The environ is shared between attempts. pyramid_retry does not copy the WSGI environ or reset it, so anything a middleware or a view stored on it persists into the next attempt. The IBeforeRetry event, which carries the request, the environ and the exception, is the place to reset such state.

  • retryable_error and last_retry_attempt are two view predicates registered when pyramid_retry is included. The first matches an exception view only for an exception that is retryable and will actually be retried; the second matches on the last attempt only, as above. The sentence in the pyramid_retry documentation example that spells the second predicate is_last_attempt=True is out of date: the registered name is last_retry_attempt (the function pyramid_retry.is_last_attempt(request) is the Python-side check).

  • pyramid_tm alone does not retry. Retries were part of pyramid_tm until 2.0; since then they need pyramid_retry, and the tm.attempts setting is gone. Older tutorials that set it describe a removed feature. Source: pyramid_retry documentation, Per-Request Attempts, Receiving Retry Notifications and Caveats

Scripts and the initialize_db Command

Work that is not a request (loading seed data, a maintenance task) still needs a session and a transaction. Pyramid offers pyramid.paster.bootstrap(config_uri), which reads an .ini file, builds the application and returns an environment with a fake request that has all the request extensions of the application, including request.tm and request.dbsession. The generated script does exactly that:

# norun  (project file bookshelf_shelves/scripts/initialize_db.py, runs against the .ini file)
import argparse
import sys

from pyramid.paster import bootstrap, setup_logging
from sqlalchemy.exc import OperationalError

from .. import models


def setup_models(dbsession):
    """Add or update models / fixtures in the database."""
    dbsession.add(models.shelf.Shelf(slug="ada", title="Ada's shelf"))


def parse_args(argv):
    parser = argparse.ArgumentParser()
    parser.add_argument("config_uri", help="Configuration file, e.g., development.ini")
    return parser.parse_args(argv[1:])


def main(argv=sys.argv):
    args = parse_args(argv)
    setup_logging(args.config_uri)
    env = bootstrap(args.config_uri)
    try:
        with env["request"].tm:
            setup_models(env["request"].dbsession)
    except OperationalError:
        print("Cannot use the database; did you run `alembic upgrade head`, and is it running?")

Source: Pyramid: Writing a Script

The key line is with env["request"].tm:. A bootstrapped request does not go through the tween, so nothing begins or ends the transaction unless the script does it. The manager created by explicit_manager is a context manager: it begins a transaction on entry, commits on a normal exit and aborts when the block raises. The project’s own initialize_bookshelf_shelves_db development.ini was run for this page and inserted its row; the same pattern, with a registry instead of an .ini file, is runnable here with pyramid.scripting.prepare:

from pyramid.scripting import prepare

app, engine = make_app()
with prepare(registry=app.registry) as env:
    request = env["request"]
    with request.tm:                          # begin; commit on exit, abort on error
        request.dbsession.add(Shelf(slug="ada", title="Ada's shelf"))
assert count(engine) == 1

try:
    with prepare(registry=app.registry) as env:
        request = env["request"]
        with request.tm:
            request.dbsession.add(Shelf(slug="never", title="Rolled back"))
            raise ValueError("the script failed half way")
except ValueError:
    pass
assert count(engine) == 1                     # the second block aborted
print("script transactions behave like request transactions")

Source: Pyramid: pyramid.scripting

pshell uses the same machinery: the setup hook of the generated pshell.py calls request.tm.begin() and exposes tm, dbsession and models, so a shell session is one open transaction that you commit with tm.commit() or discard with tm.abort(). Leaving a shell open on a production database holds locks, as any open transaction does. The commands pserve, pshell and bootstrap are covered on the command-line page of this section.

Migrations with Alembic

Alembic is the migration tool of SQLAlchemy. A revision is a Python file with upgrade() and downgrade() functions and the identifier of its parent revision; the revisions form a chain, and a table named alembic_version in the database records where the database currently is. The cookiecutter configures it in two places.

The [alembic] section of each .ini file (the same file serves the application and Alembic, which is why the alembic command is run with -c development.ini):

[alembic]
# path to migration scripts
script_location = bookshelf_shelves/alembic
file_template = %%(year)d%%(month).2d%%(day).2d_%%(rev)s

Source: Alembic: Tutorial

The doubled percent signs are the .ini escape for a literal %. The template makes the revision files sort by date (20261007_a66f218963c1.py) instead of the Alembic default of plain revision ids.

The generated env.py connects Alembic to the application’s own settings and metadata. The essential lines are:

# norun  (excerpt of bookshelf_shelves/alembic/env.py)
from alembic import context
from pyramid.paster import get_appsettings, setup_logging
from sqlalchemy import engine_from_config

from bookshelf_shelves.models.meta import Base

config = context.config
setup_logging(config.config_file_name)

settings = get_appsettings(config.config_file_name)
target_metadata = Base.metadata          # what "autogenerate" compares the database with

Source: Alembic: Tutorial, The Migration Environment

get_appsettings loads the [app:main] section, so sqlalchemy.url is read from one place; target_metadata is the metadata that autogeneration compares with the live schema, and env.py builds its own engine from the settings and runs the migrations on its own connection. It does not start the Pyramid application, so no pyramid_tm is involved: Alembic manages its transactions itself (on SQLite and MySQL, DDL is not transactional, which the log announces with "Will assume non-transactional DDL").

Creating and Applying Revisions

With the models of the previous sections, the first command creates the initial revision from the difference between an empty database and Base.metadata. The command is run with -c development.ini, as in the project README:

alembic -c development.ini revision --autogenerate -m "create shelves and books"
alembic -c development.ini upgrade head
alembic -c development.ini current

Source: Alembic: Auto Generating Migrations

The revision file it writes (here 20261007_a66f218963c1.py, trimmed) is ordinary code that you review and commit with the model change:

# norun  (generated revision script, trimmed)
revision = '...'          # a random id of 12 hexadecimal digits, also part of the file name
down_revision = None


def upgrade():
    op.create_table('shelves',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('slug', sa.String(), nullable=False),
    sa.Column('title', sa.String(), nullable=False),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_shelves')),
    sa.UniqueConstraint('slug', name=op.f('uq_shelves_slug'))
    )
    op.create_table('books',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('shelf_id', sa.Integer(), nullable=False),
    sa.Column('title', sa.String(), nullable=False),
    sa.ForeignKeyConstraint(['shelf_id'], ['shelves.id'], name=op.f('fk_books_shelf_id_shelves')),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_books'))
    )


def downgrade():
    op.drop_table('books')
    op.drop_table('shelves')

Source: Alembic: Auto Generating Migrations

The constraint names (pk_shelves, uq_shelves_slug) come from the naming convention of the models. Later model changes produce later revisions: after adding Shelf.is_public and Book.isbn to the models, the same revision --autogenerate command wrote a revision whose upgrade() is two op.add_column calls. The commands that you use every day:

Command Effect

alembic -c development.ini revision --autogenerate -m "text"

Compare the database with the models and write a revision for the difference.

alembic -c development.ini revision -m "text"

Write an empty revision, for hand-written changes such as data migrations.

alembic -c development.ini upgrade head

Apply all pending revisions. upgrade +1 applies one, upgrade <id> goes to a given revision.

alembic -c development.ini downgrade -1

Undo the last revision (and downgrade base undoes all of them).

alembic -c development.ini current, history, heads, branches, show head

Inspect the state of the database and of the revision chain.

alembic -c development.ini check

Fail when autogenerate would find pending upgrade operations: a CI guard that models and migrations agree. It prints "No new upgrade operations detected." when they do.

alembic -c development.ini stamp head

Record a revision in alembic_version without running anything, to adopt a database that was created by other means.

alembic -c development.ini upgrade head --sql

Print the SQL that would run, instead of running it (offline mode), for a DBA to review or to run by hand.

Source: Alembic: Commands

The offline mode printed, for the two revisions of this page, the CREATE TABLE statements with the constraint names and then ALTER TABLE books ADD COLUMN isbn VARCHAR; and ALTER TABLE shelves ADD COLUMN is_public BOOLEAN DEFAULT 1 NOT NULL;, each followed by an update of alembic_version. Source: Alembic: Generating SQL Scripts (a.k.a. "Offline Mode")

What Autogenerate Cannot Do

--autogenerate is a draft, not a decision. It detects added and removed tables, columns, indexes and (named) unique and foreign key constraints, and, by default, changes of column type. Changes of server default are detected only when compare_server_default is set in context.configure(), and named CHECK constraints only when enabled explicitly. It does not detect a renamed table or column (it sees a drop and an add, which would lose the data), anonymously named constraints, or some special types such as Enum on backends without a native one. Always read the script before committing it.

The common trap is the required column on a table that already has rows. A new NOT NULL column needs a value for the existing rows, and without one the migration fails (SQLite: "Cannot add a NOT NULL column with default value NULL"; other databases fail on the existing rows). That is the reason for server_default=true() on is_public in the model: the generated migration then reads server_default=sa.text('1'), nullable=False and applies cleanly to a populated table. The other safe route is a three-step revision: add the column as nullable, fill it with an op.execute statement, then alter it to NOT NULL.

SQLite is the other recurring difference: its ALTER TABLE is limited, so changes to constraints and column types need Alembic’s batch mode, which rebuilds the table, enabled with render_as_batch=True in context.configure(). The generated env.py does not enable it; add it if your development or test database is SQLite and your revisions alter existing tables. The Flask section explains batch mode, data migrations, branches and merges, multiple databases and testing migrations in more depth, and all of it applies unchanged to a Pyramid project because it is Alembic and not the framework: see Database Migrations with Flask-Migrate. Source: Alembic: Running "Batch" Migrations for SQLite and Other Databases

Checking Models Against Migrations in Code

The alembic check command has a Python counterpart, alembic.autogenerate.compare_metadata, which returns the list of differences between a live connection and the metadata. It is the basis of a test that fails when someone changes a model and forgets the revision. The block builds an old database (without books.isbn), shows the difference that autogenerate would act on, and shows that a database created from the metadata has none.

from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from sqlalchemy import create_engine, text

old = create_engine("sqlite://")
with old.begin() as connection:
    connection.execute(text("CREATE TABLE shelves (id INTEGER NOT NULL, slug VARCHAR NOT NULL, "
                            "title VARCHAR NOT NULL, is_public BOOLEAN DEFAULT 1 NOT NULL, "
                            "CONSTRAINT pk_shelves PRIMARY KEY (id), CONSTRAINT uq_shelves_slug UNIQUE (slug))"))
    connection.execute(text("CREATE TABLE books (id INTEGER NOT NULL, shelf_id INTEGER NOT NULL, "
                            "title VARCHAR NOT NULL, CONSTRAINT pk_books PRIMARY KEY (id), "
                            "CONSTRAINT fk_books_shelf_id_shelves FOREIGN KEY(shelf_id) REFERENCES shelves (id))"))

with old.connect() as connection:
    diff = compare_metadata(MigrationContext.configure(connection), Base.metadata)
print([(op, table, column.name) for op, _schema, table, column in diff])
assert [(op, table) for op, _schema, table, _column in diff] == [("add_column", "books")]

fresh = create_engine("sqlite://")
Base.metadata.create_all(fresh)
with fresh.connect() as connection:
    assert compare_metadata(MigrationContext.configure(connection), Base.metadata) == []

Source: Alembic: Autogeneration API, compare_metadata

In a project the same assertion runs against the database that the test fixture has just migrated with alembic.command.upgrade(config, "head"); the project’s test on this page is assert compare_metadata(MigrationContext.configure(connection), Base.metadata) == [] inside a test that receives the dbengine fixture, and it passed with both revisions applied.

Running Migrations in Deployment

Migrations are a deployment step, not application start-up code. Run alembic upgrade head once per release, from the release artifact, before the new application processes start taking traffic, and not from main(): with several Gunicorn workers or several replicas, every process would race to migrate. Patterns that work are a release job, an init container, or a one-shot container command, with the same environment variable for the database URL as the application; see Docker and Kubernetes for the container side. Make revisions backwards-compatible with the previous application version where you can (add the column first, deploy the code, drop the old column in a later release), so that a rollback of the code does not need a downgrade. The relational background is on SQL and relational databases.

Testing with Transaction Rollback

The cookiecutter’s tests/conftest.py turns the transaction manager into a test tool. Its pieces, in order:

  • dbengine (session scope) creates the engine, drops the tables, stamps Alembic at base, and runs alembic.command.upgrade(alembic_cfg, "head"), so the tests run on the schema that the migrations produce. It drops everything again at the end.

  • app (session scope) calls main({}, dbengine=dbengine, **app_settings): this is what the dbengine hook in includeme is for, sharing one engine with the application.

  • tm (per test) is a TransactionManager(explicit=True) that is begun and immediately doomed, so that nothing a test does can ever commit. The fixture aborts it at the end of the test.

  • dbsession (per test) is get_tm_session(session_factory, tm): a session joined to that manager.

  • testapp (per test) wraps the app in a webtest.TestApp whose environ carries tm.active = True, tm.manager = tm and app.dbsession = dbsession. pyramid_tm sees tm.active and does not start a transaction of its own, and request.dbsession returns the test’s session, so every request of the test sees the test’s uncommitted data.

The project tests written for this page (a unit test of a view and functional tests through testapp) are:

# norun  (project files tests/test_functional.py and tests/test_migrations.py)
import pytest
from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext
from sqlalchemy.exc import IntegrityError

from bookshelf_shelves.models.meta import Base


def test_post_then_get(testapp):
    testapp.post_json("/shelves", {"slug": "ada", "title": "Ada's shelf"}, status=201)
    assert testapp.get("/shelves", status=200).json[0]["slug"] == "ada"


def test_duplicate_slug_fails(testapp):
    testapp.post_json("/shelves", {"slug": "ada", "title": "A"}, status=201)
    with pytest.raises(IntegrityError):
        testapp.post_json("/shelves", {"slug": "ada", "title": "B"})


def test_rolled_back_between_tests(testapp):
    assert testapp.get("/shelves").json == []        # the shelf of the previous test is gone


def test_models_match_migrations(dbengine):
    with dbengine.connect() as connection:
        context = MigrationContext.configure(connection)
        assert compare_metadata(context, Base.metadata) == []

Source: Pyramid wiki2 tutorial: Tests

All of them passed under pytest in the generated project (with the scaffold’s test_notfound_view and a unit test of add_shelf). The block below is a runnable reduction of the same pattern, without pytest: a doomed explicit manager, a session joined to it, and a test application that is told not to manage transactions. Two requests share the uncommitted data, and nothing reaches the database.

import tempfile

with tempfile.TemporaryDirectory() as tmp:        # a file database: a second connection sees only committed rows
    app, engine = make_app(url=f"sqlite:///{tmp}/test.sqlite")
    factory = app.registry["dbsession_factory"]

    tm = transaction.TransactionManager(explicit=True)
    tm.begin()
    tm.doom()                                     # no test can commit by accident
    session = get_tm_session(factory, tm)
    testapp = TestApp(app, extra_environ={
        "tm.active": True,                        # pyramid_tm stays out of the way
        "tm.manager": tm,                         # request.tm is ours
        "app.dbsession": session,                 # request.dbsession is ours
    })

    testapp.post_json("/shelves", {"slug": "ada", "title": "A"}, status=201)
    assert [s["slug"] for s in testapp.get("/shelves").json] == ["ada"]   # visible to the next request
    assert count(engine) == 0                     # but never committed
    tm.abort()
    assert count(engine) == 0
    print("the test data disappeared with the transaction")
    session.close()
    engine.dispose()

Source: pyramid_tm documentation, Testing

Rollback-based isolation is fast and keeps tests independent, and it has limits. Code under test that depends on a real COMMIT (for example, triggers, constraints that are deferred to commit, or a second connection reading the data) behaves differently, and the savepoint limitation of SQLite hides bugs in code that uses request.tm.savepoint(). Write a handful of tests without the tm fixtures for those paths. The wider testing toolbox, including pyramid.testing, is on the testing page of this section.

Pitfalls and Security

Pitfall What happens and what to do

request.dbsession.commit() in a view

zope.sqlalchemy raises "Transaction must be committed using the transaction manager". Remove the call; the tween commits. If a view needs a partial commit, it needs a second transaction, not a call on this one.

Raw SQL with text() that writes

Silently rolled back unless mark_changed(session) is called. Prefer ORM statements, or call mark_changed.

A returned HTTPBadRequest or a 500 response

Commits, because it is a response and not an exception. Raise HTTP errors, or configure tm.commit_veto = pyramid_tm.default_commit_veto.

Using ORM objects after the request

DetachedInstanceError: the session is closed at commit. Convert to dictionaries in the view, or load what you need with selectinload/joinedload first.

Accessing request.tm or request.dbsession in a tween over pyramid_tm, a response callback after the commit, or in an exception view without tm_active

NoTransaction in explicit mode, or a session in a wrong state. Declare tm_active=True, or open your own session and transaction.

The long-running request that holds a transaction

A connection from the pool and its locks are held for the whole request. Do slow external calls outside the transaction, or exclude the endpoint with tm.activate_hook.

sqlalchemy.pool_pre_ping = false

Enables pre-ping, because the value stays a string (see Engine and Pool Settings). Omit the key.

Retried requests that send mail or call an API

The side effect runs once per attempt. Attach it to the transaction (a commit hook) or to a subscriber that runs after the commit.

sqlalchemy.echo = true in production

SQL statements and their parameters, which include personal data, reach the logs.

Creating tables with Base.metadata.create_all() at start-up

It never alters an existing table and bypasses the migration history; production schemas come from Alembic only.

pip install without pinning the SQLAlchemy series

2.1 changes the default PostgreSQL driver and autoflush rules; pin the series (or test both) when upgrading.

Source: zope.sqlalchemy on PyPI and pyramid_tm documentation

Security points specific to this layer:

  • SQL injection. SQLAlchemy binds parameters for the ORM and for select(), where() and friends. The risk returns the moment a request value is formatted into a string given to text() or to execute(): use text("... WHERE slug = :slug") with a parameters dictionary. The OWASP SQL Injection Prevention Cheat Sheet describes the defences; this site has no OWASP pages, so the reference below is external.

  • Mass assignment. Copy only the fields you allow from the request body into a model (see Using the Session in Views).

  • Credentials. Keep the database password out of .ini files in version control, and read it from the environment (The Database URL Is a Secret); use separate accounts for the application and for migrations.

  • Error output. A SQL error message can include table and column names or the statement itself. The .ini files of the cookiecutter enable the debug toolbar in development only (pyramid.includes = pyramid_debugtoolbar is in development.ini, not in production.ini), and production should render a generic error view.

  • Denial of service. Pool exhaustion is the quiet failure of an overloaded service: set pool_timeout, bound the number of threads or workers, and make slow endpoints skip the transaction rather than hold a connection. Source: OWASP: SQL Injection Prevention Cheat Sheet

Synchronous by Design

Pyramid 2.1 is a WSGI framework without async def views, and this whole stack is synchronous: the session, the transaction manager and the pool assume that a request is handled by one thread from start to end. SQLAlchemy has an asyncio extension, but it is meant for ASGI frameworks and does not combine with pyramid_tm, which has no async support. If your application needs an async database layer, that is a reason to look at FastAPI or at Tornado (see Databases & Async Drivers), not a reason to bolt asyncio onto Pyramid.

What Changed and Where Older Material Is Outdated

Older material Today Notes

from zope.sqlalchemy import ZopeTransactionExtension and sessionmaker(extension=ZopeTransactionExtension())

zope.sqlalchemy.register(session) or register(session_factory)

Renamed in zope.sqlalchemy 1.2 (2019) to ZopeTransactionEvents, with register as the supported entry point.

A global DBSession = scoped_session(sessionmaker(...)) imported by every view

request.dbsession, a reified request property

The current cookiecutter and wiki2 tutorial use the request property; the global is still seen in older tutorials and projects.

declarative_base(), Column(Integer, ...), DBSession.query(...)

DeclarativeBase, Mapped, mapped_column, select() with session.scalars()

The 2.0 style, generated by the cookiecutter 2.1-branch.

tm.attempts with pyramid_tm

pyramid_retry and retry.attempts

Removed in pyramid_tm 2.0 (2017).

The thread-local transaction.manager as request.tm

tm.manager_hook = pyramid_tm.explicit_manager

Recommended by pyramid_tm; generated by the cookiecutter.

pcreate -s alchemy, python setup.py develop, initialize_db development.ini as a bare script

The cookiecutter, pip install -e ., and the generated console script initialize_<name>_db

The pcreate scaffolds are not part of Pyramid 2.1; see the cookiecutter page of this section.

Base.metadata.create_all() in initialize_db

Alembic revisions

The older scaffold created tables from the models directly; the current one ships Alembic and a versions/ folder.

zope.sqlalchemy 3.0 and earlier on SQLAlchemy 2.0

zope.sqlalchemy 4.1

3.0 (2023) added SQLAlchemy 2.0 support; 4.0 (2025) made zope a native namespace package and excluded SQLAlchemy 2.0.32 to 2.0.35, which deadlock in its tests; 4.1 (2025-11-25) added Python 3.12 to 3.14 and dropped 3.7 to 3.9.

Source: zope.sqlalchemy on PyPI, Changes and pyramid_tm changes

pyramid_retry has not had a release since 2.1.1 (2020-03-21) and pyramid_tm since 2.6 (2024-11-14). Both are small, stable and used by the official cookiecutter, and every behaviour on this page was checked against their installed source, but when you adopt them in a new project it is reasonable to check their recent activity on PyPI and GitHub, as for any add-on (the catalogue of add-ons and their maintenance status is on the extending page of this section).

When Not to Use This Stack

The cookiecutter’s SQLAlchemy stack suits relational data with many writes per request. Other shapes exist and are covered elsewhere: an object database with the traversal tree stored in it (the cookiecutter’s zodb backend), a key-value or document store attached to the request, or a read-only service that needs no transaction at all, for which tm.activate_hook or simply not including pyramid_tm is correct. The page SQL Transactions covers what a transaction guarantees at the database level (isolation levels, locks and deadlocks, which is what makes retries necessary).

References