Correct usage of SQLAlchemy sessions-CodePudding

I am using raw SQLAlchemy in a Flask app and used to face a lot of troubles using SQLAlchemy sessions. I used to get 500 INTERNAL SERVER ERROR frequently mentioning that the last session was not closed properly or a rollback error. After identifying these issues I made modifications to my code and so far it is working great for me. But still, I get the error at times especially when my API breaks before the response. I want to know what will be the best way to use these sessions so that the commit(), rollback(), close(), etc. happen at the right time and it is generalized for all the APIs. I know that Flask-SQLAlchemy is capable of handling this automatically but I want to stick to raw SQLAlchemy.

So far the best working code for me has been -

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

    ...

I am taking care of GET, INSERT, UPDATE, DELETE in all the different routes and I am looking for a way to handle these transactions as efficiently as possible to avoid breaking the connection to the database due to any error in API.

One way is to use try-except blocks but I believe there must be a better way than mentioning try-except blocks in every route individually.

CodePudding user response：

I believe that most elegant way is by using context manager with session / transaction scope:

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

You can use it in two different way:

@app.route('/get-score', methods=['POST'])
def handle_route1():
    with Session() as session:
        with transaction_scope(session):
            ...
            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:
        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 = 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:
        return {}
    return {
       'userId': row.user_id,
       'score' : row.score
    }

CodePudding user response：

I highly recommend to read this part of the SQLAlchemy documentation.

But tl:dr you can use python context manager to control the scope of the session:

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)