SQLAlchemy-Continuum

1.6.0 · active · verified Sun Apr 12

SQLAlchemy-Continuum is a versioning and auditing extension for SQLAlchemy. It automatically creates versions for inserts, deletes, and updates of SQLAlchemy models, supports Alembic migrations, and allows reverting objects and their relations to previous states. The current version, 1.6.0, was released in January 2026, indicating an active development and maintenance cadence with recent updates for Python 3.13/3.14 and SQLAlchemy 2.0 compatibility.

Warnings

breaking SQLAlchemy-Continuum versions 1.5.0 and later have dropped support for Python 3.8. Users on Python 3.8 must use an older version of the library (e.g., <1.5.0).
Fix: Upgrade Python to 3.9+ or pin `sqlalchemy-continuum` to `<1.5.0`.
breaking SQLAlchemy-Continuum versions 1.4.0 and higher exhibit incompatibility with SQLAlchemy versions less than 2.0 due to changes in SQLAlchemy's `Connection.execute` API. This can lead to `AttributeError: 'str' object has no attribute 'is_insert'`.
Fix: Ensure `SQLAlchemy` is version `2.0` or higher when using `sqlalchemy-continuum>=1.4.0`. Alternatively, pin `sqlalchemy-continuum` to `<1.4.0` if `SQLAlchemy<2.0` is required.
gotcha Tracking changes in many-to-many relationships requires explicitly defining the association table as a SQLAlchemy model and marking *that model* with `__versioned__ = {}`. Otherwise, only changes to the primary entities, not the relationship itself, will be recorded in the `changeset`.
Fix: For many-to-many relationships, create an explicit association object model (if you don't already have one) and add `__versioned__ = {}` to it. This allows `sqlalchemy-continuum` to create a version table for the association, tracking changes to the links between entities.
gotcha For PostgreSQL native versioning, after making schema changes (e.g., adding new columns) to a versioned table, the database trigger might become outdated.
Fix: Call `sqlalchemy_continuum.dialects.postgresql.sync_trigger(connection, 'table_name_version')` to update the version trigger after schema modifications to ensure correct versioning behavior.
gotcha Inefficient querying of version history can lead to N+1 query problems, especially when traversing `.previous`/`.next` versions or relationships on version objects.
Fix: Utilize efficient querying methods like `version_at(session, primary_keys, transaction_id=...)` for specific point-in-time lookups, and `all_versions(link=True)` for batch fetching. Also, consider the `validity` strategy and ensure composite indexes on `(entity_pk, transaction_id DESC)` are active. Avoid excessive relationship traversal on version objects if possible.

Install

pip install SQLAlchemy-Continuum Install latest version

Imports

make_versioned
```
from sqlalchemy_continuum import make_versioned
```
Required to be called before model definitions to enable versioning.
version_class
```
from sqlalchemy_continuum import version_class
```
Used to get the dynamically generated version class for a given model.
parent_class
```
from sqlalchemy_continuum import parent_class
```
Used to get the original parent class from a version class.

Quickstart

This quickstart demonstrates how to enable versioning for a SQLAlchemy model. It involves calling `make_versioned()` before model definitions, adding `__versioned__ = {}` to desired models, and then calling `configure_mappers()` after all models are defined. It then shows how to create, update, and retrieve historical versions of an object, including reverting to a previous state.

import sqlalchemy as sa
from sqlalchemy import create_engine, Column, Integer, Unicode, UnicodeText
from sqlalchemy.orm import sessionmaker, declarative_base, configure_mappers
from sqlalchemy_continuum import make_versioned, version_class

# Call make_versioned() before defining models
make_versioned(user_cls=None)

Base = declarative_base()

class Article(Base):
    __tablename__ = 'article'
    __versioned__ = {}
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(Unicode(255))
    content = Column(UnicodeText)

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

# After defining all models, call configure_mappers
configure_mappers()

# Setup database and session
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# Create an article
article = Article(name='Initial Name', content='Initial Content')
session.add(article)
session.commit()
print(f"Created: {article}")

# Update the article
article.name = 'Updated Name'
session.commit()
print(f"Updated: {article}")

# Access versions
ArticleVersion = version_class(Article)
versions = session.query(ArticleVersion).filter_by(id=article.id).order_by(ArticleVersion.transaction_id).all()

print(f"\nAll versions for Article {article.id}:")
for v in versions:
    print(f"- Version (tx_id={v.transaction_id}): name='{v.name}'")

# Revert to the first version
if versions:
    first_version = versions[0]
    first_version.revert()
    session.commit()
    print(f"\nReverted to first version. Current article name: {article.name}")

session.close()

view raw JSON →