Блог

Мне нужно обеспечить транзакционность запросов flask. Есть ли способ сохранить сеанс базы данных для каждого запроса и использовать его?

Самый простой способ — создать простой extension для инициализации сеанса и зафиксировать изменения после успешного запроса. Вот пример:

 from datetime import datetime

from flask import Flask, current_app, _app_ctx_stack
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base, DeferredReflection
from sqlalchemy import Column, String, Integer

# Base model without connection(see: DBExtension.connect()) and a few demo models
Base = declarative_base(cls=DeferredReflection)


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String())


class Log(Base):
    __tablename__ = 'log'
    id = Column(Integer, primary_key=True, autoincrement=True)
    text = Column(String())


# extension for getting db session
class DBExtension:
    def __init__(self, app: Flask = None):
        self.app = app
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        app.teardown_appcontext(self.teardown)

    def connect(self):
        # init engine and set into metadata
        engine = create_engine(current_app.config['DB_CONNECTION'], echo=True)
        Session = sessionmaker(bind=engine)
        session = Session()
        # create_all - just for demo(init structure...)
        Base.metadata.create_all(engine)
        Base.metadata.bind = engine
        Base.prepare(engine)
        return session

    def teardown(self, exception):
        ctx = _app_ctx_stack.top
        if hasattr(ctx, 'session'):
            ctx.session.close()

    @property
    def session(self):
        # connect or get active connection
        ctx = _app_ctx_stack.top
        if ctx is not None:
            if not hasattr(ctx, 'session'):
                ctx.session = self.connect()
            return ctx.session


app = Flask(__name__)
app.config.update({'DB_CONNECTION': 'sqlite:///test.db'})  # sqlite for demo
db = DBExtension(app)


@app.teardown_request
def teardown_request_func(error=None):
    # request processing completed(before requests functions, endpoints and other logic)
    if not error:
        # one more record and commit all changes in one transaction
        db.session.add(Log(text='{dt}. teardown_request'.format(dt=datetime.utcnow())))
        try:
            db.session.commit()
        except Exception as e:
            print(e)
            # do something


@app.before_request
def before_request_func():
    # add a new record before request
    db.session.add(Log(text='{dt}. before request'.format(dt=datetime.utcnow())))


@app.route('/user/<name>')
def user_route(name: str):
    # do something with db session
    db.session.add(Log(text='saving user {user}'.format(user=name)))
    user = User(name=name)
    db.session.add(user)
    # session.add(newModel) or session.delete(oneMoreModel) etc.
    return 'done'  # return without commit. see: teardown_request_func


if __name__ == '__main__':
    app.run(host='localhost', port=5000, debug=True)
  

Запустите сервер, откройте http://localhost:5000/user/DavidBoshton и просмотрите logs( BEGIN-...-COMMIT ).