Сохранение в 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)

Сохранение времени в базе данных

Очень часто бывает, что практически все знают, как надо делать правильно, но при этом всё равно постоянно делают неправильно. Один из таких случаев — сохрание времени в базе данных. Понятно, что на персональном блоге вполне можно обойтись наивных подходом, не учитывающим перевод времени. Но для круглосуточно работающих приложений строгой системы отчётности вроде биллинга это неприемлемо.

Я не буду здесь рассматривать все возможные варианты корректной работы со временем. Покажу лишь насколько просто можно реализовать самый распространённый вариант — хранение в базе в UTC — на примере SQLAlchemy. Для этого достаточно определить новый тип колонки:

from sqlalchemy import types
from dateutil.tz import tzutc
from datetime import datetime

class UTCDateTime(types.TypeDecorator):

    impl = types.DateTime

    def process_bind_param(self, value, engine):
        if value is not None:
            return value.astimezone(tzutc()).replace(tzinfo=None)

    def process_result_value(self, value, engine):
        if value is not None:
            return datetime(value.year, value.month, value.day,
                            value.hour, value.minute, value.second,
                            value.microsecond, tzinfo=tzutc())

Теперь вы можете сохранять время с произвольной зоной, все преобразования будут сделаны автоматически. Но сохранить время без зоны не получится — метод astimezone() выбросит исключение ValueError, что позволит избежать случайных ошибок.

Логгинг в базу или борьба с рекурсией

Как-то понадобилось мне сделать логгинг в базу для модуля logging. Сразу видна очевидная проблема: внутри такого обработчика идёт сохранения в базу некоторыми принятыми в проекте средствами, которые сами используют logging. В результате обработчик рекусивно будет вызывать себя и зацикливаться. Понятно, что можно отбросить привычные средства и сделать либо логгинг своими средствами без использования стандартного пакета logging, либо в базу писать низкоуровневым кодом, который logging не использует. Но это всё не интересно и ведёт за собой неудобства в использовании.

Первая мысль была выставлять флаг в обработчике перед записью в базу и снимать его после. Но это без ухищрений не будет работать в многопоточном приложении, а ведь есть ещё и обработчики сигналов. А нельзя ли определить, что обработчик был вызван из самого себя? Оказывается, можно — с помощью средств работы со стеком интерпретатора в модуле inspect. Достаточно убедиться, чтобы текущего метода не было в стеке вызовов. Сам фрейм не подходит для сравнения, так как он создаётся новый на каждый вызов, но можно сравнивать объекты кода. Первая версия на базе inspect.stack() оказалась достаточно медленной, а ведь запись в лог — это то, что используется постоянно. Дело в том, что эта функция подготавливает много лишней информации, которая нам не нужна. Зато проход по стеку "вручную" оказался достаточно быстрым, примерно на 2 порядка быстрее варианта с inspect.stack():

def isRecursive():
   '''Returns whether it's recursive call of caller function.'''
   frame = inspect.currentframe().f_back
   try:
       code = frame.f_code
       while True:
           frame = frame.f_back
           if frame is None:
               break
           if frame.f_code is code:
               return True
       return False
   finally:
       del frame

Сам обработчик достаточно прост (здесь storage — произвольное хранилище, в моём случае оно сделано на базе SQLAlchemy):

class DBServiceLogHandler(logging.Handler):

   def __init__(self, storage):
       logging.Handler.__init__(self)
       self._storage = storage

   def emit(self, record):
       # We can't check this in filter() method, since recursive call is from
       # emit, not filter.
       if isRecursive():
           return
       traceback = None
       if record.exc_info:
           # It's used by logging to cache
           if not record.exc_text:
               record.exc_text = self.formatException(record.exc_info)
           traceback = record.exc_text
       entry = LogEntry(name=record.name, level=record.levelno,
                        message=record.getMessage(), traceback=traceback)
       try:
           self._storage.store(entry)
       except self._storage.Error:
           logger.exception('Error logging to DB (%s):', record.getMessage())
       self._storage.clear()