How do I maintain multi-table integrity across multiple selects in SQLAlchemy in Pyramid?

Question

I'm trying to build a Pyramid application. I started with the SQLAlchemy scaffolding. I'm running into a issue and I'm wondering what the best way is to address it. In one of my views I need to select a lot of rows from two unrelated tables. I need to make sure no rows were inserted into the second table between the time I selected the rows from the first table and the time I selected the rows from the second table.

I have three models, Node, Test, and Tasking. Both Nodes and Tests have quite a bit of metadata. Given a list of Nodes and a list of Tests, a global list of Taskings can be created. For example, we could have three Nodes, a, b, and c and two Tests "we need one node to do task P" and "we need two nodes to do task Q".

From that information, three Tasks should be created. For example:

"Node a should do task P"
"Node b should do task Q"
"Node c should do task Q"

Now, I'm attempting to provide a REST API for this. The vast majority of time clients will be requesting the list of Tasks, so that needs to be fast. However, sometimes a client might add a Node or a Test. When that happens, I need the entire list of Tasks to be regenerated.

Here's a rough example:

@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)

I'm using the default Pyramid SQLAlchemy scaffolding. So, each request auto starts a transaction. So, if _update_tasking is called from one request (say add_node), then the new node will be added to the local DBSession, and querying for all Nodes and Tests in _update_tasking will return that new element. In addition, deleting all existing Taskings and adding the newly computed ones is also safe.

I have two problems:

What happens if a new row is added into the Tests table between the time I get my list of nodes and my list of tests in _update_taskings? In my real world production system these selects are close together but not right next to each other. There is the possibility of a race condition.
How do I ensure two requests that will update the Taskings don't overwrite each other? For example, imagine if our existing system had one Node and one Test. Two requests come in at the same, one to add a Node and one to add a Test. Even if problem #1 wasn't an issue and I knew that each request's pair of selects represented "a single instance of time in the database", there's still the problem of one request overriding the other. If the first request finishes first with now two Nodes and one Test, the second request will still be selecting the old data (potentially) and will generate a list of Taskings with one Node and two Tests.

So, what's the best way to deal with this? I'm using SQLite for development and PostgreSQL in production, but I'd like a database agnostic solution. I'm not worried about other applications accessing this database. My REST API will be the only access mechanism. Should I put a lock around any of requests that mutate the database (adding a Node or a Test)? Should I lock the database somehow?

Thanks for any help!

rkhayrov rkhayrov · Accepted Answer · 2013-08-21T08:09:19

Using serializable transaction isolation level should prevent both problems. If one transaction modifies the data that can affect results of previous reads in another transaction, there is a serialization conflict. Only one transaction wins, all others are aborted by the database to be restarted by the client. SQLite does this by locking the whole database, PostgreSQL employs much more complex mechanism (see docs for details). Unfortunately, there is no portable sqlalchemic way to catch serialization anomaly and retry. You need to write DB-specific code to reliably distinguish it from other errors.

I've put up a sample program with two threads concurrently modifying the data (a very basic reproduction of your scheme), running into conflicts and retrying:

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

With Pyramid transaction middleware and Zope transaction manager in use that would be even easier. After catching serialization error, instead of retrying manually, raise TransientError and middleware will retry the whole request up to tm.attempts (in the paster config) times.

from transaction.interfaces import TransientError

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

You can even write your own middleware sitting below pyramid_tm in the stack that will catch serialization errors and translate them to transient errors transparently.

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

How do I maintain multi-table integrity across multiple selects in SQLAlchemy in Pyramid?

1 Answers