python - 如何在 Pyramid 中的 SQLAlchemy 中跨多个选择维护多表完整性？

Question

我正在尝试构建一个 Pyramid 应用程序。我从 SQLAlchemy 脚手架开始。我遇到了一个问题，我想知道解决它的最佳方法是什么。在我的一个观点中，我需要从两个不相关的表中选择很多行。我需要确保在我从第一个表中选择行和从第二个表中选择行之间没有行插入到第二个表中。

我有三个模型，Node、Test 和 Tasking。 Nodes 和 Tests 都有相当多的元数据。给定一个 Nodes 列表和一个 Tests 列表，可以创建一个全局 Taskings 列表。例如，我们可以有三个 Nodes、a、b 和 c 以及两个 Tests“我们需要一个节点来执行任务P”和“我们需要两个节点来执行任务Q”。

根据该信息，应创建三个 Tasks。例如:

1. “节点 a 应该执行任务 P”
2. “节点 b 应该执行任务 Q”
3. “节点 c 应该执行任务 Q”

现在，我正在尝试为此提供一个 REST API。绝大多数时间客户端将请求 Tasks 列表，因此需要快速。但是，有时客户端可能会添加一个 Node 或一个 Test。发生这种情况时，我需要重新生成整个 Tasks 列表。

这是一个粗略的例子:

@view_config(route_name='list_taskings')
def list_taskings(request):
    return DBSession.Query(Tasking).all()

@view_config(route_name='add_node')
def add_node(request):
    DBSession.add(Node())
    _update_taskings()

@view_config(route_name='add_test')
def add_test(request):
    DBSession.add(Test())
    _update_taskings()

def _update_taskings():
    nodes = DBSession.query(Node).all()
    tests = DBSession.query(Test).all()

    # Process...

    Tasking.query.delete()
    for t in taskings:
        DBSession.add(t)

我正在使用默认的 Pyramid SQLAlchemy 脚手架。因此，每个请求都会自动启动一个事务。因此，如果 _update_tasking 从一个请求中调用(比如 add_node)，那么新节点将被添加到本地 DBSession，并查询_update_tasking 中的所有 Nodes 和 Tests 都将返回该新元素。此外，删除所有现有的 Taskings 并添加新计算的任务也是安全的。

我有两个问题:

1. 如果在我获取节点列表和测试列表<之间将新行添加到Tests 表中，会发生什么情况 在 _update_taskings 中？在我的现实世界生产系统中，这些选择靠得很近，但彼此并不紧邻。存在竞争条件的可能性。

2. 如何确保将更新 Taskings 的两个请求不会相互覆盖？例如，假设我们现有的系统有一个 Node 和一个 Test。两个请求同时进来，一个是添加一个Node，一个是添加一个Test。即使问题 #1 不是问题并且我知道每个请求的一对选择代表“数据库中的单个时间实例”，但仍然存在一个请求覆盖另一个请求的问题。如果第一个请求首先完成，现在有两个 Nodes 和一个 Test，第二个请求仍将(可能)选择旧数据并生成 列表使用一个节点和两个测试的任务。

那么，处理这个问题的最佳方法是什么？我在开发中使用 SQLite，在生产中使用 PostgreSQL，但我想要一个与数据库无关的解决方案。我不担心其他应用程序访问这个数据库。我的 REST API 将是唯一的访问机制。我是否应该锁定任何改变数据库的请求(添加 Node 或 Test)？我应该以某种方式锁定数据库吗？

感谢您的帮助!

Answer 1

使用 serializable事务隔离级别应该防止这两个问题。如果一个事务修改了可能影响另一个事务中先前读取结果的数据，则存在序列化冲突。只有一个事务获胜，所有其他事务都被数据库中止并由客户端重新启动。 SQLite 通过锁定整个数据库来做到这一点，PostgreSQL 使用更复杂的机制(详见 docs)。不幸的是，没有可移植的 sqlalchemic 方法来捕获序列化异常并重试。您需要编写特定于数据库的代码以可靠地将其与其他错误区分开来。

我已经提出了一个示例程序，其中有两个线程同时修改数据(您的方案的非常基本的复制)，遇到冲突并重试:

https://gist.github.com/khayrov/6291557

使用 Pyramid 事务中间件和 Zope 事务管理器会更容易。捕获序列化错误后，不是手动重试，而是引发 TransientError 并且中间件将重试整个请求，直到 tm.attempts(在粘贴配置中)次。

from transaction.interfaces import TransientError

class SerializationConflictError(TransientError):
    def __init__(self, orig):
        self.orig = orig

您甚至可以编写位于堆栈中 pyramid_tm 下方的自己的中间件，以捕获序列化错误并将它们透明地转换为 transient 错误。

def retry_serializable_tween_factory(handler, registry):

    def retry_tween(request):
        try:
            return handler(request)
        except DBAPIError, e:
            orig = e.orig
            if getattr(orig, 'pgcode', None) == '40001':
                raise SerializationConflictError(e)
            elif isinstance(orig, sqlite3.DatabaseError) and \
                orig.args == ('database is locked',):
                raise SerializationConflictError(e)
            else:
                raise

    return retry_tween