正确使用SQLAlchemy会话

Question

我在Flask应用程序中使用原始SQLAlchemy，并且使用SQLAlchemy会话时遇到了很多麻烦。我过去常常得到500 INTERNAL SERVER ERROR，经常提到最后一个会话没有正确关闭或回滚错误。在确定了这些问题之后，我修改了我的代码，到目前为止，它对我来说工作得很好。但是，我仍然有时会得到错误，特别是当我的API在响应之前中断时。我想知道使用这些会话的最佳方式是什么，以便commit(),rollback(),close()等在正确的时间发生，并且它适用于所有api。我知道Flask-SQLAlchemy能够自动处理这个问题，但我想坚持使用原始的SQLAlchemy。

到目前为止对我来说最好的工作代码是-

from flask import Flask
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind = mysql_engine())
db_session = Session()
@app.route('/get-score', methods=['POST'])
def handle_route1():
...
row = db_session.query(DB_Model_1) 
.filter(DB_Model_1.user_id == user_id) 
.all()
row = row[0] if row else None
if not row:
db_session.close()
return {}
db_session.close()
return {
'userId': row.user_id,
'score' : row.score
}
@app.route('/insert-score', methods=['POST'])
def handle_route2():
...
@app.route('/update-score', methods=['POST'])
def handle_route3():
...
@app.route('/delete-score', methods=['POST'])
def handle_route3():
...

我在所有不同的路由中照顾GET,INSERT,UPDATE,DELETE，我正在寻找一种方法来尽可能有效地处理这些事务，以避免由于API中的任何错误而中断与数据库的连接。

一种方法是使用try-except但是我相信一定有比在每条路由中单独提到try-except块更好的方法。

Answer 1

我认为最优雅的方法是在会话/事务范围内使用上下文管理器:

from contextlib import contextmanager
@contextmanager
def transaction_scope(session, close_at_exit=False):
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
if close_at_exit:
session.close()

有两种用法:

1。

@app.route('/get-score', methods=['POST'])
def handle_route1():
with Session() as session:
with transaction_scope(session):
...
row = session.query(DB_Model_1) 
.filter(DB_Model_1.user_id == user_id) 
.all()
row = row[0] if row else None
if not row:
return {}
return {
'userId': row.user_id,
'score' : row.score
}

@app.route('/get-score', methods=['POST'])
def handle_route1():
session = Session()
with transaction_scope(session, close_at_exit=True):
...
row = session.query(DB_Model_1) 
.filter(DB_Model_1.user_id == user_id) 
.all()
row = row[0] if row else None
if not row:
return {}
return {
'userId': row.user_id,
'score' : row.score
}

Answer 2

我强烈建议您阅读SQLAlchemy文档的这一部分。

但是tl:dr你可以使用python上下文管理器来控制会话的范围:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine('postgresql://scott:tiger@localhost/')
Session = sessionmaker(engine)
with Session.begin() as session:
session.add(some_object)