Python/Flask/SQLAlchemy 实例列表中的数据在页面重新加载后仍然存在
Python/Flask/SQLAlchemy data in instance list persisting after page reloads
项目结构
app/
-- entrypoint.py
-- database.py
-- models/
---- person.py
---- filtering_class.py
-- pages/
---- routes.py
问题中包含的文件:
entrypoint.py
from flask import Flask
import models
app = Flask(__name__, instance_relative_config=True)
database.py
from flask import _app_ctx_stack
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
SQLALCHEMY_DATABASE_URI = 'sqlite:///persons.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
engine = create_engine(
SQLALCHEMY_DATABASE_URI,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=engine
)
session = scoped_session(
SessionLocal,
scopefunc=_app_ctx_stack.__ident_func__
)
Base = declarative_base()
routes.py
from flask import render_template, request
from entrypoint import app
from database import session
from models import Person
@app.route('/person')
def person():
response_id = request.args.get('id')
person = session.query(Person).filter_by(id=response_id).first()
person = Person.get_others(person=person)
return render_template('person.html',
person=person)
person.py
from database import Base, session
from models import FilteringClass
class Person(Base):
__tablename__ = 'persons'
id = Column(Integer, primary_key=True)
first_name = Column(String(80), nullable=True)
last_name = Column(String(80), nullable=True)
@staticmethod
def get_persons(filtering_class=None):
query = session.query(Person.id,
Person.first_name,
Person.last_name)
if filtering_class:
query = filter_class.apply_filters()
results = query.all()
@staticmethod
def get_others(person):
# Here lies the problem
filtering_class = FilteringClass()
# After first call, len(filtering_class.custom_expressions) == 0 (expected), after second call len(...) already == 1, etc
filtering_class.add_custom_expression(Person.id != person.id)
return Person.get_persons(filtering_class=filtering_class)
filtering_class.py
class FilteringClass(object):
def __init__(self,
custom_expressions=[]):
self.custom_expressions = custom_expressions
def apply_filters(self, query):
# Import Person here to avoid circular import issues in person.py
from models.person import Person
if self.custom_expressions:
for exp in self.custom_expressions:
query = query.filter(exp)
return query
def add_custom_expression(self, expression):
self.custom_expressions.append(expression)
描述
FilteringClass 用于过滤传递的查询参数。它有一个方法供 class 的用户添加他们自己的 BinaryExpressions
以在调用 FilteringClass.apply_filters()
时应用。
这里的目标是检索所有 与发起页面的人 不同的 Person
使用 FilteringClass
请求排除具有相同 ID 的 Person
个对象。
问题
预期的行为是在每次请求时实例化一个新的 FilteringClass
(请参阅 Person.get_others
--> filtering_class = FilteringClass()
)。
到那时,预计 filtering_class
实例中的内部 custom_expressions
数组将是空的,如其构造函数中所定义。
但是,每次重新加载与 /person
路由相关的页面并创建 filtering_class
的实例时,它的 custom_expressions
数组已经填充了先前添加的自定义表达式。这意味着在每次重新加载页面时,filtering_class.custom_expressions
都会增长而不会回到空状态。
我试过的
- 在过滤传递的查询后直接使用
self.custom_expressions = []
将 custom_expressions
重置为空。
- 在
/person
端点返回页面模板之前调用 session.close()
。
- 在
/person
端点返回页面模板之前调用 session.commit()
(但我认为无论如何我都不应该为 SELECT 语句提交任何内容)。
抱歉这么久了 post,我试着把所有有用的东西都包括进来(但如果我应该添加任何东西,请告诉我)。
更新:解决方案:
根据@larsks 的评论,问题 与 SQLAlchemy 无关 而是与具有默认可变参数的 Python 陷阱有关(参见:https://docs.python-guide.org/writing/gotchas/#mutable-default-arguments ).
解决该问题所需要做的就是:
- 更改
FilteringClass
' 构造函数自:
def __init__(self,
custom_expressions=[]):
self.custom_expressions = custom_expressions
至:
def __init__(self,
custom_expressions=None):
self.custom_expressions = custom_expressions
- 将
add_custom_expression
方法更改为:
def add_custom_expression(self, expression):
self.custom_expressions.append(expression)
至:
def add_custom_expression(self, expression):
if self.custom_expressions is None:
self.custom_expressions = []
self.custom_expressions.append(expression)
项目结构
app/
-- entrypoint.py
-- database.py
-- models/
---- person.py
---- filtering_class.py
-- pages/
---- routes.py
问题中包含的文件:
entrypoint.py
from flask import Flask
import models
app = Flask(__name__, instance_relative_config=True)
database.py
from flask import _app_ctx_stack
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
SQLALCHEMY_DATABASE_URI = 'sqlite:///persons.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
engine = create_engine(
SQLALCHEMY_DATABASE_URI,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=engine
)
session = scoped_session(
SessionLocal,
scopefunc=_app_ctx_stack.__ident_func__
)
Base = declarative_base()
routes.py
from flask import render_template, request
from entrypoint import app
from database import session
from models import Person
@app.route('/person')
def person():
response_id = request.args.get('id')
person = session.query(Person).filter_by(id=response_id).first()
person = Person.get_others(person=person)
return render_template('person.html',
person=person)
person.py
from database import Base, session
from models import FilteringClass
class Person(Base):
__tablename__ = 'persons'
id = Column(Integer, primary_key=True)
first_name = Column(String(80), nullable=True)
last_name = Column(String(80), nullable=True)
@staticmethod
def get_persons(filtering_class=None):
query = session.query(Person.id,
Person.first_name,
Person.last_name)
if filtering_class:
query = filter_class.apply_filters()
results = query.all()
@staticmethod
def get_others(person):
# Here lies the problem
filtering_class = FilteringClass()
# After first call, len(filtering_class.custom_expressions) == 0 (expected), after second call len(...) already == 1, etc
filtering_class.add_custom_expression(Person.id != person.id)
return Person.get_persons(filtering_class=filtering_class)
filtering_class.py
class FilteringClass(object):
def __init__(self,
custom_expressions=[]):
self.custom_expressions = custom_expressions
def apply_filters(self, query):
# Import Person here to avoid circular import issues in person.py
from models.person import Person
if self.custom_expressions:
for exp in self.custom_expressions:
query = query.filter(exp)
return query
def add_custom_expression(self, expression):
self.custom_expressions.append(expression)
描述
FilteringClass 用于过滤传递的查询参数。它有一个方法供 class 的用户添加他们自己的 BinaryExpressions
以在调用 FilteringClass.apply_filters()
时应用。
这里的目标是检索所有 与发起页面的人 不同的 Person
使用 FilteringClass
请求排除具有相同 ID 的 Person
个对象。
问题
预期的行为是在每次请求时实例化一个新的 FilteringClass
(请参阅 Person.get_others
--> filtering_class = FilteringClass()
)。
到那时,预计 filtering_class
实例中的内部 custom_expressions
数组将是空的,如其构造函数中所定义。
但是,每次重新加载与 /person
路由相关的页面并创建 filtering_class
的实例时,它的 custom_expressions
数组已经填充了先前添加的自定义表达式。这意味着在每次重新加载页面时,filtering_class.custom_expressions
都会增长而不会回到空状态。
我试过的
- 在过滤传递的查询后直接使用
self.custom_expressions = []
将custom_expressions
重置为空。 - 在
/person
端点返回页面模板之前调用session.close()
。 - 在
/person
端点返回页面模板之前调用session.commit()
(但我认为无论如何我都不应该为 SELECT 语句提交任何内容)。
抱歉这么久了 post,我试着把所有有用的东西都包括进来(但如果我应该添加任何东西,请告诉我)。
更新:解决方案:
根据@larsks 的评论,问题 与 SQLAlchemy 无关 而是与具有默认可变参数的 Python 陷阱有关(参见:https://docs.python-guide.org/writing/gotchas/#mutable-default-arguments ).
解决该问题所需要做的就是:
- 更改
FilteringClass
' 构造函数自:
def __init__(self,
custom_expressions=[]):
self.custom_expressions = custom_expressions
至:
def __init__(self,
custom_expressions=None):
self.custom_expressions = custom_expressions
- 将
add_custom_expression
方法更改为:
def add_custom_expression(self, expression):
self.custom_expressions.append(expression)
至:
def add_custom_expression(self, expression):
if self.custom_expressions is None:
self.custom_expressions = []
self.custom_expressions.append(expression)