区块链技术博客
www.b2bchain.cn

Flask SQLAlchemy-仅在当前会话中设置expire_on_commit = False – python程序员分享

D0b2wT.gif

本文介绍了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***** 

Flask_SQLAlchemy无法像SQLAlchemy那样反映数据库 – python

我正在尝试使用python-Flask将Foreigner Database连接到Flask_SQLALchemy应用我到处都看过,包括FLASK_SQLALCHEMY official doc 过去4天里,我一直在Internet上到处寻找FLASK_SQLALCHEMY中没有ORM库的任何教程,但运气不佳我一直在查看SQLAlchemy Reflecti…

SQLalchemy数据库列应采用哪种模型来包含数据数组? – python

因此,我正在尝试建立一个数据库,该数据库的行将经常修改。例如,每小时,我想向数据库的特定部分添加一个数字。因此,如果将self.checkmarks输入等于3的数据库,最好的方法是用增加的数字更新数据库的此部分以使self.checkmarks现在等于3, 2?我尝试将列建立为db.Array,但出现属性错误: AttributeError:“ SQLAlc…

将非线程SQLAlchemy代码与Flask-SQLAlchemy集成 – python

我有一个Python模块UserManager,它负责与用户管理相关的所有事情-用户,组,权限,身份验证。通过在构造函数中传递SQLAlchemy引擎参数的主类提供对这些资产的访问。需要引擎来进行表类映射(使用映射器对象)并发出会话。这是在app模块中建立gobal变量的方式:class UserManager: def __init__(self, db)…

python flask如何从URL获取路由ID – python

我正在用烧瓶。我有一个网址,例如:http://example.com/page/page_id我想知道如何从路线中的网址获取page_id部分。我希望我可以设计出一些方法,例如:@route('/page/page_id') def page(page_id): pageid = page_id 参考方案 这非常简单-在尖括号之间传递p…

flask sqlalchemy-根据数据库中的列动态生成数据模型 – python

有没有一种方法可以根据Flask SQLAlchemy的数据库表中的列动态生成数据库模型?我有一个应用程序来显示数据库表中的数据,但是列名有时会更改并破坏我的应用程序。我希望有一种方法可以根据数据库中的实际列名动态​​生成数据模型。我目前明确声明所有列,如下所示。class MyDbModel(db.Model): __tablename__ = &#039…

部分转自互联网,侵权删除联系

赞(0) 打赏
部分文章转自网络,侵权联系删除b2bchain区块链学习技术社区 » Flask SQLAlchemy-仅在当前会话中设置expire_on_commit = False – python程序员分享
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

b2b链

联系我们联系我们