python - SQLAlchemy session 的正确用法

Question

我在 Flask 应用程序中使用原始 SQLAlchemy，过去使用 SQLAlchemy session 时遇到很多麻烦。我曾经经常收到 500 INTERNAL SERVER ERROR，提到上次 session 未正确关闭或回滚错误。在确定这些问题后，我修改了我的代码，到目前为止它对我来说效果很好。但是，我有时还是会收到错误消息，尤其是当我的 API 在响应之前中断时。我想知道使用这些 session 的最佳方式是什么，以便 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 block ，但我相信一定有比在每个路由中单独提及 try-except block 更好的方法。

Answer 1

我认为最优雅的方法是使用具有 session /事务范围的上下文管理器:

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
       }