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