Zope/SQLAlchemy Transaction Integration

4.1 · active · verified Thu Apr 16

zope.sqlalchemy provides minimal integration between SQLAlchemy and Zope's transaction management system. It acts as a data manager, unifying various approaches to allow SQLAlchemy ORM sessions to participate in Zope transactions, often used in web frameworks like Pyramid. The library is currently at version 4.1, released in November 2025, and maintains an active release schedule.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to set up `zope.sqlalchemy` with a SQLAlchemy engine and a scoped session. It defines a simple `User` model, creates an in-memory SQLite database, registers the `DBSession` factory, and then uses a Zope transaction to add and retrieve data. Note the explicit `transaction.commit()` and `transaction.abort()`, and `DBSession.remove()` after each operation to properly manage the session lifecycle.

import os
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, scoped_session, declarative_base
from zope.sqlalchemy import register
import transaction

# 1. Define a simple SQLAlchemy model
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"

# 2. Create an engine (using an in-memory SQLite for example)
# Use an environment variable for a real database DSN, e.g., POSTGRES_DSN='postgresql://user:pass@host/db'
database_url = os.environ.get('TEST_DSN', 'sqlite:///:memory:')
engine = create_engine(database_url)

# 3. Create tables (only if they don't exist)
Base.metadata.create_all(engine)

# 4. Create a SQLAlchemy session factory and scope it
# Scoped sessions are important for web applications or multi-threaded environments
Session = sessionmaker(bind=engine)
DBSession = scoped_session(Session)

# 5. Register the scoped session factory with zope.sqlalchemy
# This integrates its commits/rollbacks with Zope transactions
register(DBSession)

# --- Usage Example ---

# Get a session from the factory (it's thread-local/scoped)
session = DBSession()

try:
    # Add a new user
    new_user = User(name='Alice')
    session.add(new_user)

    # The transaction manager handles committing/rolling back the session
    transaction.commit()
    print(f"Added user: {new_user}")

    # After commit, the session might be closed/invalidated by default
    # Attempting to access new_user might raise DetachedInstanceError if not reloaded
    # print(new_user.name) # This might fail if session is closed.

except Exception as e:
    transaction.abort()
    print(f"Transaction aborted due to error: {e}")
finally:
    # Ensure the session is removed from the scope, especially in web requests
    DBSession.remove()

# Start a new transaction for another operation
session = DBSession()

try:
    # Query users
    users = session.query(User).all()
    print(f"Current users: {users}")
    transaction.commit()
except Exception as e:
    transaction.abort()
    print(f"Query transaction aborted: {e}")
finally:
    DBSession.remove()

view raw JSON →