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

如何只为Flask-SQLAlchemy中的当前会话设置选项expire_on_commit=False

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

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

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

我尝试了db.session.expire_on_commit = False,但似乎没有任何效果。

参考方案

expire_on_commitsqlalchemy.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

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时,您都在使用相同的基础会话。这是与上述相同的示例,但是使用了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创建的所有会话都将选项设置为
错误,相反,我只想为一个会话设置它。

鉴于Flask-SQLAlchemy每个请求仅创建一个会话的事实,我认为这意味着您在处理请求时有时希望将会话设为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*****

可以看出,提交后,将重新发出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*****

45码