SQLAlchemy - 为复杂的混合创建表达式 属性
SQLAlchemy - create expression for complex hybrid property
我在 MyModel
中有这样一个 hybrid property
:
@hybrid_property
def state(self):
states = [dummy.state_id for dummy in self.dummies.all()]
if all(state == "DUMMY" for state in states):
return State.query.get("DUMMY").text
if all((state == "FAKE" or state == "DUMMY") for state in states):
return State.query.get("FAKE").text
return State.query.get("INVALID").text
我想在我的资源中这样查询它:
valid_text = State.query.get("FAKE").text
return data_layer.model.query.filter_by(state=valid_text) # Where data_layer.model is MyModel
但是我得到一个空数组。只需 data_layer.model.query.all()
即可获取数据,因此逻辑有效。
我知道我可能需要为我的 属性 创建一个 expression
,但我发现的每个示例都用于更简单的用例。
我试过这个:
@state.expression
def state(cls):
states = [dummy.state_id for dummy in self.dummies.all()]
all_dummies = all(state == "DUMMY" for state in states)
all_fakes_or_dummies = all(
(state == "FAKE" or state == "DUMMY") for state in states
)
dummy_text = State.query.get("DUMMY").text
fake_text = State.query.get("FAKE").text
invalid_text = State.query.get("INVALID").text
return case(
[
(
all_dummies,
dummy_text,
),
(
all_fakes_or_dummies,
fake_text,
),
],
else_=invalid_text,
)
但是我的资源现在returnssqlalchemy.exc.ArgumentError: Ambiguous literal: False. Use the 'text()' function to indicate a SQL expression literal, or 'literal()' to indicate a bound value.
我想知道如何才能正确实现此 python 逻辑以与 SQLAlchemy 兼容,我想这一定是问题所在。另外我想知道在混合 属性 中制作如此复杂的逻辑是否是一个好习惯。
在您的情况下,基于 NOT EXIST
的查询构建应该会为您提供所需的结果,同时生成易于阅读的查询。
和你一样,我试图将支票分解成一些独立的块,结果如下:
class MyModel(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String)
# ...
# Hybrid Properties
@hybrid_property
def has_only_dummies(self):
# this is not optimal as need to reiterate over all objects
states = [dummy.state_id for dummy in self.dummies.all()]
return all(state == "DUMMY" for state in states)
@hybrid_property
def has_only_dummies_or_fake(self):
# this is not optimal as need to reiterate over all objects
states = [dummy.state_id for dummy in self.dummies.all()]
return all(state in ("DUMMY", "FAKE") for state in states)
@hybrid_property
def state(self):
# could reuse other hybrid properties here, but it is not efficient at all
res = None
states = [dummy.state_id for dummy in self.dummies.all()]
if all(state == "DUMMY" for state in states):
res = "DUMMY"
elif all((state == "FAKE" or state == "DUMMY") for state in states):
res = "FAKE"
else:
res = "INVALID"
return res
# Hybrid Expressions
@has_only_dummies.expression
def has_only_dummies(cls):
subq = (
exists()
.where(Dummy.model_id == cls.id)
.where(~Dummy.state_id.in_(["DUMMY"]))
).correlate(cls)
return select([case([(subq, False)], else_=True)]).label("only_dum")
@has_only_dummies_or_fake.expression
def has_only_dummies_or_fake(cls):
subq = (
exists()
.where(Dummy.model_id == cls.id)
.where(~Dummy.state_id.in_(["DUMMY", "FAKE"]))
).correlate(cls)
return select([case([(subq, False)], else_=True)]).label("only_dum_or_fak")
@state.expression
def state(cls):
return db.case(
[
(cls.has_only_dummies, "DUMMY"),
(cls.has_only_dummies_or_fake, "FAKE"),
],
else_="INVALID",
)
在这种情况下,您可以构建如下查询,包括过滤:
q = session.query(MyModel, MyModel.has_only_dummies, MyModel.has_only_dummies_or_fake, MyModel.state)
q = session.query(MyModel, MyModel.state)
q = session.query(MyModel).filter(MyModel.state != "INVALID")
MyModel.state
并不是您想要的(它是 state_id
而不是 text
),但是获取文本是最重要的一步易于实施,以防您确实需要它。
我在 MyModel
中有这样一个 hybrid property
:
@hybrid_property
def state(self):
states = [dummy.state_id for dummy in self.dummies.all()]
if all(state == "DUMMY" for state in states):
return State.query.get("DUMMY").text
if all((state == "FAKE" or state == "DUMMY") for state in states):
return State.query.get("FAKE").text
return State.query.get("INVALID").text
我想在我的资源中这样查询它:
valid_text = State.query.get("FAKE").text
return data_layer.model.query.filter_by(state=valid_text) # Where data_layer.model is MyModel
但是我得到一个空数组。只需 data_layer.model.query.all()
即可获取数据,因此逻辑有效。
我知道我可能需要为我的 属性 创建一个 expression
,但我发现的每个示例都用于更简单的用例。
我试过这个:
@state.expression
def state(cls):
states = [dummy.state_id for dummy in self.dummies.all()]
all_dummies = all(state == "DUMMY" for state in states)
all_fakes_or_dummies = all(
(state == "FAKE" or state == "DUMMY") for state in states
)
dummy_text = State.query.get("DUMMY").text
fake_text = State.query.get("FAKE").text
invalid_text = State.query.get("INVALID").text
return case(
[
(
all_dummies,
dummy_text,
),
(
all_fakes_or_dummies,
fake_text,
),
],
else_=invalid_text,
)
但是我的资源现在returnssqlalchemy.exc.ArgumentError: Ambiguous literal: False. Use the 'text()' function to indicate a SQL expression literal, or 'literal()' to indicate a bound value.
我想知道如何才能正确实现此 python 逻辑以与 SQLAlchemy 兼容,我想这一定是问题所在。另外我想知道在混合 属性 中制作如此复杂的逻辑是否是一个好习惯。
在您的情况下,基于 NOT EXIST
的查询构建应该会为您提供所需的结果,同时生成易于阅读的查询。
和你一样,我试图将支票分解成一些独立的块,结果如下:
class MyModel(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String)
# ...
# Hybrid Properties
@hybrid_property
def has_only_dummies(self):
# this is not optimal as need to reiterate over all objects
states = [dummy.state_id for dummy in self.dummies.all()]
return all(state == "DUMMY" for state in states)
@hybrid_property
def has_only_dummies_or_fake(self):
# this is not optimal as need to reiterate over all objects
states = [dummy.state_id for dummy in self.dummies.all()]
return all(state in ("DUMMY", "FAKE") for state in states)
@hybrid_property
def state(self):
# could reuse other hybrid properties here, but it is not efficient at all
res = None
states = [dummy.state_id for dummy in self.dummies.all()]
if all(state == "DUMMY" for state in states):
res = "DUMMY"
elif all((state == "FAKE" or state == "DUMMY") for state in states):
res = "FAKE"
else:
res = "INVALID"
return res
# Hybrid Expressions
@has_only_dummies.expression
def has_only_dummies(cls):
subq = (
exists()
.where(Dummy.model_id == cls.id)
.where(~Dummy.state_id.in_(["DUMMY"]))
).correlate(cls)
return select([case([(subq, False)], else_=True)]).label("only_dum")
@has_only_dummies_or_fake.expression
def has_only_dummies_or_fake(cls):
subq = (
exists()
.where(Dummy.model_id == cls.id)
.where(~Dummy.state_id.in_(["DUMMY", "FAKE"]))
).correlate(cls)
return select([case([(subq, False)], else_=True)]).label("only_dum_or_fak")
@state.expression
def state(cls):
return db.case(
[
(cls.has_only_dummies, "DUMMY"),
(cls.has_only_dummies_or_fake, "FAKE"),
],
else_="INVALID",
)
在这种情况下,您可以构建如下查询,包括过滤:
q = session.query(MyModel, MyModel.has_only_dummies, MyModel.has_only_dummies_or_fake, MyModel.state)
q = session.query(MyModel, MyModel.state)
q = session.query(MyModel).filter(MyModel.state != "INVALID")
MyModel.state
并不是您想要的(它是 state_id
而不是 text
),但是获取文本是最重要的一步易于实施,以防您确实需要它。