python - Flask SQLAlchemy - 仅为当前 session 设置 expire_on_commit=False

Question

如何设置选项expire_on_commit=False仅适用于 Flask-SQLAlchemy 中的当前 session ？

我可以在 SQLAlchemy 的初始化时设置选项对象:

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

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

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

Answer 1

expire_on_commit是 sqlalchemy.orm.session.Session 的参数类(class)。

获取 Session 的首选方法例如，是通过 sqlalchemy.orm.session.sessionmaker 类(class)。 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

一个 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.session处理 request 时您正在使用相同的基础 session 。这是与上面相同的示例，但使用 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) 完全相同的结果在前面的例子中。

all sessions created by Flask-SQLAlchemy will have the option set to False, instead i want to set it only for one session.

鉴于 Flask-SQLAlchemy 只为每个请求创建一个 session ，我认为这意味着您有时需要一个 session 到 expire_on_commit。有时在处理请求时不会。

实现此目的的一种方法是使用 context manager暂时转 expire_on_commit离开:

from contextlib import contextmanager

@contextmanager
def no_expire():
    s = db.session()
    s.expire_on_commit = False
    try:
        yield
    finally:
        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*****

可以看到commit之后，重新发出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*****