Сохранение в SQLAlchemy под контролем

Большинство задач, для которых используется SQLAlchemy — это веб-приложения, отличающиеся небольшим количеством действий, выполняемых на один запрос, короткими транзакциями. И для этих целей типовая схема работы, расписанная в документации, подходит очень хорошо. Но работа с базами данных нужна не только в веб-приложениях, и даже в веб-приложениях иногда есть отдельные процессы с более сложными операциями.

Типовая схемы работы предполагает накопление некоторого количества изменений и вызов метода flush() у сессии, который сохраняет все изменения в базе. А теперь представьте, что будет, если в ходе работы на одной из итераций мы получаем исключение, мы это исключение обрабатываем и продолжаем работу? Вполне резонно, что часть ("ошибочных") накопленных изменений должна пропасть, то есть не попасть в базу. Но ведь метод flush() предполагает сохранения именно всех изменений. Конечно, мы можем очистить сессию и произвести инициализацию заново — достаточно неудобно, да и зачем снова загружать данные, которые не могли измениться? Кто-то резонно заметит, что в метод flush() можно передать список объектов для сохранения. Да, это именно то, что нужно. Только следует понимать, что в этом случае сохраняться будут только эти объекты, но не объекты, которые от них зависят, то есть cascade rules перестают работать. В итоге мы не можем использовать autoflush=True и должны самостоятельно отслеживать каскадные правила при сохранении. Аналогично не стоит использовать transactional=True, так как в этом случае транзакция открывается сразу же после закрытия предыдущей, и при длительной работе без commit()-ов могут возникать значительные замедления в работе базы данных.

Ещё одна неприятная особенность есть у SessionTransaction. Используя другие библиотеки для работы с базами данных я привык, что можно определить метод с некоторой транзакцией, а затем вызывать его из другого метода, в котором к исходным действиям добавляются ещё какие-то, и всё это, конечно, в одной общей транзакции. Но дело в том, что сессионные транзакции в SQLAlchemy не могут быть вложенными. На самом деле всё гораздо хуже, они могут быть вложенными, но результат будет отличным от ожидаемого: транзакция будет закрыта уже при при вызове commit() внутренней транзакции. Проблема решается использованием объекта транзакции для соединения, который работает как нужно.

Подытожу всё сказанное в классе Storage (недостающие методы не представляют сложности в реализации):

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


class Storage(object):

    def __init__(self, dbURL):
        self._engine = create_engine(dbURL)
        self._conn = self._engine.connect()
        self._session = sessionmaker(bind=self._conn, autoflush=False,
                                     transactional=False)()

    def transaction(self):
        return self._conn.begin()

    def store(self, obj):
        with self.transaction():
            self._session.save_or_update(obj)
            from sqlalchemy.orm.session import _cascade_iterator
            cascaded = [o for o, m in _cascade_iterator('save-update', obj)]
            self._session.flush([obj]+cascaded)