SQLAlchemy版本控制有内置吗

本文目录导读：

1. 使用 version_id_col（内置乐观锁）
2. 使用事件监听器实现简单历史记录
3. 使用 SQLAlchemy-Continuum（推荐）
4. 使用 Alembic + 自定义实现
5. 选择建议

是的，SQLAlchemy 本身没有内置完整的版本控制功能（不像 Django 的 django-reversion 那样开箱即用）,但提供了构建版本控制所需的底层工具和模式。

以下是实现 SQLAlchemy 版本控制的几种常见方案,从简单到完整：

使用 version_id_col（内置乐观锁）

SQLAlchemy 的 ORM 内置了简单的版本控制，主要用于并发控制（防止丢失更新）,而不是历史记录。

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Article(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)= Column(String)
    content = Column(String)
    version_id = Column(Integer, default=1)  # 版本字段
    __mapper_args__ = {
        'version_id_col': version_id  # 声明版本列
    }

特点：

• 自动递增版本号
• 更新时会检查版本号是否匹配（避免脏写）
• 不保存历史记录，仅用于并发控制

使用事件监听器实现简单历史记录

通过 before_update 事件将旧版本保存到历史表：

from sqlalchemy import event, create_engine
from sqlalchemy.orm import sessionmaker
class ArticleHistory(Base):
    __tablename__ = 'articles_history'
    id = Column(Integer, primary_key=True)
    article_id = Column(Integer)  # 原记录ID= Column(String)
    content = Column(String)
    version = Column(Integer)
    changed_at = Column(DateTime, default=datetime.utcnow)
@event.listens_for(Article, 'before_update')
def receive_before_update(mapper, connection, target):
    # 获取旧数据
    old_article = session.query(Article).get(target.id)
    # 保存到历史表
    history = ArticleHistory(
        article_id=old_article.id,
        title=old_article.title,
        content=old_article.content,
        version=old_article.version_id
    )
    session.add(history)

使用 SQLAlchemy-Continuum（推荐）

这是专门为 SQLAlchemy 设计的版本控制扩展,功能最完整：

pip install sqlalchemy-continuum

from sqlalchemy_continuum import make_versioned
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 初始化
make_versioned()
class Article(Base):
    __tablename__ = 'articles'
    __versioned__ = {}  # 标记为需要版本控制
    id = Column(Integer, primary_key=True)= Column(String)
    content = Column(String)
# 使用
article = session.query(Article).get(1)
article.title = "新标题"
session.commit()  # 自动保存版本历史
# 查询历史版本
versions = article.versions.all()
for version in versions:
    print(version.title, version.transaction.issued_at)

功能：

• 自动创建历史表（如 articles_version）
• 支持回滚到任意版本
• 支持事务关联
• 支持关联对象的版本控制

使用 Alembic + 自定义实现

如果项目已经使用 Alembic 进行数据库迁移,可以结合使用：

# 手动实现版本表
class Version(Base):
    __tablename__ = 'versions'
    id = Column(Integer, primary_key=True)
    record_type = Column(String)  # 记录类型（表名）
    record_id = Column(Integer)   # 记录ID
    data = Column(JSON)           # 完整的数据快照
    version = Column(Integer)
    created_at = Column(DateTime)
    user_id = Column(Integer)     # 谁做的修改
# 配合 Alembic 迁移脚本管理版本表结构变化

选择建议

对于大多数需要历史版本追踪的场景，SQLAlchemy-Continuum 是最省事的选择，它提供了类似 Django django-reversion 的体验。