Python + MongoDB 文档版本控制

Question

我正在开发一个应用程序，可在我工作的公司内部用作项目/任务跟踪器。玩弄 MongoDB atm。我想到了以下伪模式：

task
    _id
    name
    project
    initial_notes
    versions
        number
        versions
            version_1
                worker
                status
                date(if submitted)
                review_notes(if rejected)
                reply_on(if accepted/rejected)
            (version_n)(if any)

我遇到的问题是对任务进行版本控制。我已经阅读了许多可能的方法，但我一直没有完全理解它们。我读了一些我喜欢 here 的东西，并且真的很喜欢 mongoid 的做法 versioning

想起来更好，我宁愿有这样的东西

task
    _id
    versions
        number_of_versions: 3
        current_version
            version_no: 3
            worker: bob
            status: accepted
        old_versions
            version
                version_no: 2
                worker: bob

我想仅在显示任务集合时显示最新版本，并且我想在输入特定任务的详细信息页面时显示特定任务的所有版本。这种结构行得通吗？如果是，为了实现我的需要，需要运行哪些查询？

提前感谢您抽出宝贵的时间阅读本文并回答。 状态：拒绝 版本 版本号：1 工人：史密斯 状态：拒绝

Answer 1

是的，为什么不呢。那个方案会奏效。另外，你有没有考虑过这样的事情：

task 
    ...
    versions = [       # MongoDB array 
        {   version_id 
            worker
            status
            date(if submitted)
            review_notes(if rejected)
            reply_on(if accepted/rejected)
        },
        { version_id : ... }, 
        ... 

可能的版本插入查询：

tasks.update( { # a query to locate a particular task}, 
              { '$push' : { 'versions', { # new version } } } )

注意，在这种情况下，从版本数组中检索最后一个版本是由程序完成的，而不是 Mongo。

Answer 2

您也可以考虑这样的模型：

task
    _id
    ...
    old_versions [
        {
            retired_on
            retired_by
            ...
        }
    ]

这样做的好处是您的当前数据始终处于顶层（您无需显式跟踪当前版本，文档就是它自己的当前版本），并且您可以通过获取当前版本轻松跟踪历史记录, 删除old_versions 字段，并将$pushing 到数据库中的old_versions 字段。

由于您似乎希望最小化网络 IO，这也可以让您轻松避免在不需要时加载 old_versions：

> db.tasks.find({...}, {old_versions: 0})

您还可以更高级地存储已更改字段的旧版本列表。这需要在您的应用程序层中编写更精细的代码，如果您不希望有很多修订或这些文档非常大，则可能没有必要。

Answer 3

我正在处理同样的问题，这就是我创建 HistoricalCollection 的原因：

https://pypi.org/project/historical-collection/

就像一个普通的集合一样工作，除了一些额外的方法：

• patch_one()
• patch_many()
• find_revisions()
• latest()

用法示例：

from historical_collection.historical import HistoricalCollection
from pymongo import MongoClient
class Users(HistoricalCollection):
    PK_FIELDS = ['username', ]  # <<= This is the only requirement

# ...

users = Users(database=db)

users.patch_one({"username": "darth_later", "email": "darthlater@example.com"})
users.patch_one({"username": "darth_later", "email": "darthlater@example.com", "laser_sword_color": "red"})

list(users.revisions({"username": "darth_later"}))

# [{'_id': ObjectId('5d98c3385d8edadaf0bb845b'),
#   'username': 'darth_later',
#   'email': 'darthlater@example.com',
#   '_revision_metadata': None},
#  {'_id': ObjectId('5d98c3385d8edadaf0bb845b'),
#   'username': 'darth_later',
#   'email': 'darthlater@example.com',
#   '_revision_metadata': None,
#   'laser_sword_color': 'red'}]