Избегание кода обработки сеанса шаблона в функциях sqlalchemy

У меня есть приложение python, которое имеет множество небольших функций доступа к базе данных, используя sqlalchemy. Я стараюсь избегать большого количества кода обработки сессий с помощью шаблонов вокруг этих функций.

У меня есть несколько функций, которые выглядят примерно так:

def get_ticket_history(Session, ticket_id):
    s = Session()
    try:
        rows = s.query(TicketHistory)\
                .filter(TicketHistory.ticket_fk==ticket_id)\
                .order_by(TicketHistory.id.desc()).all()
        s.commit()
        return rows
    except:
        s.rollback()
        raise
    finally:
        s.close()

Я пытаюсь реорганизовать эти функции, но не уверен, что у меня есть лучший подход. Самое лучшее, что у меня есть, это следующее:

def execute(Session, fn, *args, **kwargs):
    s = Session()
    try:
        ret = fn(s, *args, **kwargs)
        s.commit()
        return ret
    except:
        s.rollback()
        raise
    finally:
        s.close()

def get_ticket_history(self, ticket_id):
    def sql_fn(s):
        return s.query(TicketHistory)\
                .filter(TicketHistory.ticket_fk==ticket_id)\
                .order_by(TicketHistory.id.desc()).all()
    return execute(self.sentinel_session, sql_fn)

Есть ли лучший или более идиоматический способ сделать это? Возможно, используя декоратор?

Спасибо, Джон

Ответы

from contextlib import contextmanager

@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

class ThingOne(object):
    def go(self, session):
        session.query(FooBar).update({"x": 5})

class ThingTwo(object):
    def go(self, session):
        session.query(Widget).update({"q": 18})

def run_my_program():
    with session_scope() as session:
        ThingOne().go(session)
        ThingTwo().go(session)