在 sqlalchemy 函数中避免样板会话处理代码 [英] Avoiding boilerplate session handling code in sqlalchemy functions
问题描述
我有一个使用 sqlalchemy 的 python 应用程序,它有很多小型数据库访问功能.我试图避免在这些函数周围有大量样板会话处理代码.
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?
谢谢,乔恩
推荐答案
SQLAlchemy 文档提供了一种使用上下文管理器执行此操作的可能方法.
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-何时提交并何时关闭
为了完整起见,复制此处的代码片段:
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()
这个 session_scope
现在可以干净地使用,而无需重复样板.
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)
这篇关于在 sqlalchemy 函数中避免样板会话处理代码的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!