Avoiding boilerplate session handling code in sqlalchemy functions

Question

I have a python application which has lots of small database access functions, using sqlalchemy. I'm trying to avoid having lots of boilerplate session handling code around these functions.

I have numerous functions that look something like this:

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

I am trying to refactor these functions, but not sure I have the best approach yet. The best I currently have is the following:

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) 

Is there a better or more idiomatic way of doing this? Perhaps using a decorator?

Thanks, Jon

Answer 1

The SQLAlchemy docs present a possible way of doing this with context managers.

http://docs.sqlalchemy.org/en/latest/orm/session_basics.html#when-do-i-construct-a-session-when-do-i-commit-it-and-when-do-i-close-it

Copying the code snippet here for completeness:

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

This session_scope can be used cleanly without repeating the boiler plate now.

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)