Flask SQLAlchemy-仅在当前会话中设置expire_on_commit = False

Question

如何expire_on_commit=False仅在Flask-SQLAlchemy中为当前会话设置选项？

我可以通过以下方式在SQLAlchemy对象的初始化上设置该选项：

db = SQLAlchemy(app, session_options={"expire_on_commit": False})

但是通过这种方式，Flask-SQLAlchemy创建的所有会话都将选项设置为False，而我只想为一个会话设置它。

我尝试过，db.session.expire_on_commit = False但是似乎没有任何效果。

Answer 1

expire_on_commit是sqlalchemy.orm.session.Session该类的参数。

获取Session实例的首选方法是通过sqlalchemy.orm.session.sessionmaker类。的实例sessionmaker配置了用于创建Session实例的设置。例如：

>>> from sqlalchemy import create_engine
>>> from sqlalchemy.orm import sessionmaker
>>> engine = create_engine('sqlite:///:memory:')
>>> Session = sessionmaker(bind=engine)
>>> type(Session)
<class 'sqlalchemy.orm.session.sessionmaker'>
>>> session = Session()
>>> type(session)
<class 'sqlalchemy.orm.session.Session'>

因此，调用sessionmaker实例将返回一个Session实例。

使用此配置，每次调用sessionmaker实例时，每次都会返回一个新 Session实例。

>>> session1 = Session()
>>> session2 = Session()
>>> session1 is session2
False

A scoped_session更改了上述行为：

>>> from sqlalchemy.orm import scoped_session
>>> Session = scoped_session(sessionmaker(bind=engine))
>>> type(Session)
<class 'sqlalchemy.orm.scoping.scoped_session'>
>>> session1 = Session()
>>> session2 = Session()
>>> session1 is session2
True

这就是Flask-SQLAlchemy在“幕后”使用的（以及为什么@CodeLikeBeaker的注释将您定向到Session API的原因是有效的）。这意味着，每次调用db.sessiona时，request您都在使用相同的基础会话。这是与上述相同的示例，但使用了Flask-SQLAlchemy扩展名。

>>> type(db.session)
<class 'sqlalchemy.orm.scoping.scoped_session'>
>>> session1 = db.session()
>>> session2 = db.session()
>>> session1 is session2
True

请注意，type(db.session)在此示例中产生的结果与type(Session)上一个示例完全相同。

Flask-SQLAlchemy创建的所有会话都将选项设置为False，而我只想为一个会话设置它。

考虑到Flask-SQLAlchemy每个请求仅创建一个会话的事实，我认为这意味着您expire_on_commit在处理请求时有时想要一个会话，有时不希望。

一种可以实现的方法是使用上下文管理器暂时expire_on_commit关闭：

from contextlib import contextmanager

@contextmanager
def no_expire():
    s = db.session()
    s.expire_on_commit = False
    yield
    s.expire_on_commit = True

这是我的测试模型：

class Person(db.Model):

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(16))

配置日志记录以查看SQLAlchemy正在做什么：

import logging
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
logging.basicConfig(level=logging.INFO)

创建一些测试数据：

db.drop_all()
db.create_all()
names = ('Jane', 'Tarzan')
db.session.add_all([Person(name=n) for n in names])
db.session.commit()

这是我用于测试的功能：

def test_func():
    # query the db
    people = Person.query.all()
    # commit the session
    db.session.commit()
    # iterate through people accessing name to see if sql is emitted
    for person in people:
        print(f'Person is {person.name}')
    db.session.rollback()

我无需上下文管理器即可运行一次测试功能：

test_func()

这是标准输出：

INFO:sqlalchemy.engine.base.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.base.Engine:SELECT person.id AS person_id, person.name AS person_name
FROM person
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:COMMIT
INFO:sqlalchemy.engine.base.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.base.Engine:SELECT person.id AS person_id, person.name AS person_name
FROM person
WHERE person.id = %(param_1)s
INFO:sqlalchemy.engine.base.Engine:{'param_1': 1}
*****Person is Jane*****
INFO:sqlalchemy.engine.base.Engine:SELECT person.id AS person_id, person.name AS person_name
FROM person
WHERE person.id = %(param_1)s
INFO:sqlalchemy.engine.base.Engine:{'param_1': 2}
*****Person is Tarzan*****

可以看出，在提交之后，将重新发出sql以刷新对象属性。

并一旦使用上下文管理器：

db.session.rollback()
with no_expire():
    test_func()

这是上下文管理器的标准输出：

INFO:sqlalchemy.engine.base.Engine:ROLLBACK
INFO:sqlalchemy.engine.base.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.base.Engine:SELECT person.id AS person_id, person.name AS person_name
FROM person
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:COMMIT
*****Person is Jane*****
*****Person is Tarzan*****