paper-trail-py 0.1.0

Modern Python library for tracking changes to SQLAlchemy models

🧾 PaperTrail - Python

现代化的 SQLAlchemy 模型版本追踪库 - 为你的数据库变更提供完整的审计日志。

受 Ruby PaperTrail 启发，使用 SQLAlchemy 2.0+、uv 和现代 Python 工具链构建。

✨ 特性

• 🎯 简单易用 - 一个装饰器即可启用版本追踪
• 🔍 完整历史 - 追踪所有 Create、Update、Delete 操作
• 👤 Whodunnit - 记录谁做了什么改变
• 🔄 版本恢复 - 轻松回滚到任意历史版本
• 🔗 事务分组 - 关联相关的多个变更
• ⚡ 高性能 - 支持批量操作和异步查询
• 🎨 类型安全 - 完整的类型提示支持
• 🧪 全面测试 - 70%+ 测试覆盖率
• 🗄️ 多数据库支持 - SQLite、PostgreSQL、MySQL

📦 安装

# 使用 uv（推荐）
uv pip install paper-trail-py

# 使用 pip
pip install paper-trail-py

# PostgreSQL 支持（推荐用于生产环境）
uv pip install "paper-trail-py[postgresql]"

# 异步支持
uv pip install "paper-trail-py[async]"

# MySQL 支持
uv pip install "paper-trail-py[mysql]"

🚀 快速开始

1. 定义模型并启用追踪

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase
from paper_trail import track_versions

class Base(DeclarativeBase):
    pass

@track_versions()
class Article(Base):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    content = Column(String(1000))

2. 自动记录变更

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from paper_trail import set_whodunnit

# PostgreSQL (推荐用于生产环境)
engine = create_engine('postgresql://user:pass@localhost/db')

# 或使用 SQLite (开发环境)
# engine = create_engine('sqlite:///app.db')

session = Session(engine)

# 设置操作者
set_whodunnit('[e-mail]')

# 创建文章
article = Article(title="Hello World", content="First post")
session.add(article)
session.commit()  # ✅ 自动创建版本记录

# 更新文章
article.title = "Hello Python"
session.commit()  # ✅ 自动记录变更

3. 查询历史版本

from paper_trail import VersionQuery

# 获取文章的所有版本
versions = (
    VersionQuery(session)
    .for_model(Article, article.id)
    .order_by_time(ascending=False)
    .all()
)

for v in versions:
    print(f"{v.event} by {v.whodunnit} at {v.created_at}")
    print(f"Changes: {v.changeset}")

4. 恢复历史版本

from paper_trail import reify_version

# 获取之前的版本
old_version = versions[1]

# 恢复
restored_article = reify_version(session, old_version, Article, commit=True)
print(restored_article.title)  # 恢复到旧标题

5. 查看变更详情

# 使用 changeset 属性获取变更（返回元组）
for version in versions:
    if version.event == 'update':
        for field, (old_val, new_val) in version.changeset.items():
            print(f"{field}: {old_val} → {new_val}")
    elif version.event == 'create':
        for field, new_val in version.changeset.items():
            print(f"{field}: {new_val}")

📚 核心功能详解

🎨 版本追踪装饰器

# 基础用法
@track_versions()
class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String)

# 忽略特定字段
@track_versions(ignore={'updated_at', 'view_count'})
class Article(Base):
    # ...

# 仅追踪特定字段
@track_versions(only={'title', 'status'})
class Document(Base):
    # ...

# 注意：装饰器使用 SQLAlchemy Session 事件系统
# - before_flush: 捕获变更前的状态
# - after_flush_postexec: 创建版本记录

👤 上下文管理 (Whodunnit)

from paper_trail import set_whodunnit, whodunnit

# 全局设置
set_whodunnit('[e-mail]')

# 使用上下文管理器
with whodunnit('[e-mail]'):
    article.title = "New Title"
    session.commit()

🔗 事务分组

from paper_trail.context import transaction_group

# 将多个变更关联到一个事务
with transaction_group() as tx_id:
    article1.title = "Update 1"
    article2.title = "Update 2"
    session.commit()
    
    # 所有版本记录会有相同的 transaction_id

🔍 强大的查询 API

from paper_trail import VersionQuery
from datetime import datetime, timedelta

# 查询特定用户的操作
user_versions = (
    VersionQuery(session)
    .by_user('[e-mail]')
    .after(datetime.now() - timedelta(days=7))
    .all()
)

# 查询特定事务的所有变更
tx_versions = (
    VersionQuery(session)
    .by_transaction(tx_id)
    .all()
)

# 组合查询
recent_updates = (
    VersionQuery(session)
    .for_model_type(Article)
    .by_event('update')
    .between(start_date, end_date)
    .limit(10)
    .all()
)

⚡ 性能优化

from paper_trail.performance import bulk_track_changes, cleanup_old_versions

# 批量追踪变更
articles = session.query(Article).filter(...).all()
count = bulk_track_changes(
    session,
    articles,
    Article,
    event='update',
    whodunnit='[e-mail]'
)

# 清理旧版本（保留 30 天）
deleted = cleanup_old_versions(session, days=30, model_class=Article)
print(f"Deleted {deleted} old versions")

🌊 异步支持

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from paper_trail.async_support import get_versions_async, reify_version_async

engine = create_async_engine('postgresql+asyncpg://...')

async with AsyncSession(engine) as session:
    # 异步查询版本
    versions = await get_versions_async(session, Article, article_id)
    
    # 异步恢复
    restored = await reify_version_async(session, versions[0], Article)

🗄️ 数据库模式

CREATE TABLE versions (
    id SERIAL PRIMARY KEY,                -- PostgreSQL 使用 SERIAL
    item_type VARCHAR(255) NOT NULL,      -- 模型表名
    item_id VARCHAR(255) NOT NULL,        -- 记录 ID
    event VARCHAR(50) NOT NULL,           -- create/update/destroy
    whodunnit VARCHAR(255),               -- 操作者
    transaction_id VARCHAR(255),          -- 事务分组 ID
    object JSONB,                         -- 完整快照 (PostgreSQL 使用 JSONB)
    object_changes JSONB,                 -- 变更增量 (PostgreSQL 使用 JSONB)
    created_at TIMESTAMP NOT NULL,
    
    INDEX idx_item_lookup (item_type, item_id),
    INDEX idx_transaction_lookup (transaction_id, created_at),
    INDEX idx_whodunnit_lookup (whodunnit, created_at)
);

重要提示：

• 推荐使用 PostgreSQL 作为生产数据库（支持 JSONB 类型）
• SQLite 仅适合开发和测试环境
• object_changes 在数据库中存储为 JSON 对象，使用 version.changeset 属性获取 Python 元组格式

🛠️ 开发指南

环境设置

# 克隆仓库
git clone https://github.com/yunshang/paper-trail-py.git
cd paper-trail-py

# 安装 uv（如果还没安装）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 安装开发依赖
make dev-install

# 设置 PostgreSQL 测试数据库
createdb paper_trail_test
export DATABASE_URL="postgresql://user:password@localhost/paper_trail_test"

# 安装 pre-commit hooks
pre-commit install

运行测试

# 运行所有测试
make test

# 带覆盖率报告
make test-cov

# 类型检查
make type-check

# Lint 检查
make lint

# 格式化代码
make format

# 注意：测试使用事务隔离和表重建策略确保测试独立性

项目结构

paper-trail-py/
├── src/
│   └── paper_trail/
│       ├── __init__.py          # 公共 API
│       ├── models.py            # Version 数据模型（含 changeset 属性）
│       ├── decorators.py        # @track_versions（基于 Session 事件）
│       ├── context.py           # Whodunnit 和事务管理
│       ├── query.py             # 版本查询 API
│       ├── reify.py             # 版本恢复功能
│       ├── serializers.py       # 对象序列化（使用原生 SQL 查询）
│       ├── config.py            # 全局配置管理
│       ├── async_support.py     # 异步操作支持
│       └── performance.py       # 批量操作和性能优化
├── tests/
│   ├── conftest.py              # 测试配置（PostgreSQL + 事务隔离）
│   ├── test_decorators.py       # 装饰器测试
│   ├── test_query.py            # 查询 API 测试
│   ├── test_reify.py            # 版本恢复测试
│   ├── test_context.py          # 上下文管理测试
│   ├── test_serializers.py      # 序列化测试
│   └── test_performance.py      # 性能测试
├── docs/
│   ├── ARCHITECTURE.md          # 架构设计文档
│   ├── QUICKSTART.md            # 快速入门指南
│   ├── API.md                   # API 参考
│   └── CONTRIBUTING.md          # 贡献指南
├── pyproject.toml               # uv 项目配置
├── Makefile                     # 开发命令
├── .pre-commit-config.yaml      # Pre-commit hooks
└── README.md

🏗️ 技术架构

Session 事件系统

本库使用 SQLAlchemy 2.0 的 Session 事件而非 Mapper 事件：

# before_flush - 捕获变更前的状态
@event.listens_for(Session, 'before_flush')
def before_flush_handler(session, flush_context, instances):
    # 使用原生 SQL 查询获取旧值，避免 identity map 问题
    old_values = session.execute(
        select(table).where(table.c[pk] == pk_value)
    ).first()

# after_flush_postexec - 创建版本记录
@event.listens_for(Session, 'after_flush_postexec')
def after_flush_handler(session, flush_context):
    # 批量插入版本记录
    session.bulk_insert_mappings(Version, pending_versions)

变更检测策略

1. 捕获阶段 (before_flush): 在 flush 前查询数据库获取旧值
2. 比较阶段: 对比新旧值生成 changeset
3. 存储阶段 (after_flush_postexec): 批量创建版本记录
4. 访问阶段: 使用 version.changeset 将 JSON 数组转换为 Python 元组

📊 配置选项

from paper_trail import configure

configure(
    enabled=True,                        # 全局开关
    default_ignore_fields={'updated_at'},  # 默认忽略字段
    store_object_snapshot=True,          # 存储完整快照
    store_object_changes=True,           # 存储变更增量
    batch_insert_threshold=100,          # 批量插入阈值
)

🎯 成功标准

✅ 功能完整性

• 版本追踪装饰器 (基于 Session 事件)
• Version 数据模型 (含 changeset 属性)
• 版本查询 API (VersionQuery)
• 版本恢复 (Reify)
• 上下文管理 (Whodunnit)
• 事务分组
• 自定义序列化 (原生 SQL 查询)
• 配置管理
• 异步支持
• 性能优化 (批量操作)

✅ 代码质量

• 类型提示覆盖 100%
• 测试覆盖率 70%+ (17 个测试用例)
• Ruff + Black 格式化
• MyPy 类型检查通过
• Pre-commit hooks

✅ 文档

• 完整的 README
• 架构设计文档 (ARCHITECTURE.md)
• 快速入门指南 (QUICKSTART.md)
• API 参考 (API.md)
• 贡献指南 (CONTRIBUTING.md)

🔧 技术栈

• Python 3.10+
• SQLAlchemy 2.0+ (Session Events API)
• PostgreSQL (推荐) / SQLite (开发)
• uv (包管理器)
• pytest (测试框架)
• ruff + black (代码格式化)
• mypy (类型检查)

💡 最佳实践

1. 使用 changeset 而非 object_changes

# ✅ 推荐：使用 changeset（返回元组）
for field, (old, new) in version.changeset.items():
    print(f"{field}: {old} → {new}")

# ❌ 不推荐：使用 object_changes（返回列表）
# object_changes 直接来自 JSON 字段，值是列表而非元组

2. PostgreSQL 用于生产环境

# ✅ 生产环境
engine = create_engine('postgresql://user:pass@localhost/db')

# ❌ 仅开发环境
# engine = create_engine('sqlite:///app.db')

3. 设置适当的忽略字段

# 忽略自动更新的时间戳字段
@track_versions(ignore={'updated_at', 'modified_at', 'last_sync'})
class Model(Base):
    ...

4. 使用事务分组关联变更

from paper_trail.context import transaction_group

with transaction_group() as tx_id:
    # 多个相关变更
    user.email = "[e-mail]"
    user.profile.bio = "Updated bio"
    session.commit()
    
    # 查询这个事务的所有变更
    versions = VersionQuery(session).by_transaction(tx_id).all()

🤝 贡献

欢迎贡献！请查看 CONTRIBUTING.md 了解详情。

📄 许可证

MIT License - 详见 LICENSE 文件

🙏 致谢

• 灵感来源：PaperTrail (Ruby)
• 使用工具：SQLAlchemy、uv

📮 联系方式

• Issue Tracker: https://github.com/yunshang/paper-trail-py/issues
• 文档: https://paper-trail-py.readthedocs.io

---

使用 ❤️ 和 🐍 构建