在sqlalchemy函数中避免样板会话处理代码

Question

我有一个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) 

是否有这样做的更好或更地道的方式？也许使用装饰器？

感谢， 乔恩

Answer 1

morphyn的使用上下文管理的建议是很好的。您可以通过将contextlib.contextmanager修饰器应用于与第一个get_ticket_history非常相似的功能来创建上下文管理器，将其替换为try和yield语句之间的代码，并将其重命名为transaction。 PEP 343有一个近似相同的例子。

然后，使用该语句管理器和with语句重新实现get_ticket_history。它看起来像的SQLAlchemy已经提供了该功能，但是，作为法begin：

http://docs.sqlalchemy.org/en/rel_0_8/orm/session.html#autocommit-mode

Answer 2

的SQLAlchemy的文档呈现与上下文管理这样的一个可行的办法。

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

复制代码段是为了完整性：

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

这session_scope可以干净而不现在重复锅炉板一起使用。

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)