英文:
Delete SQLAlchemy query using row_number()
问题
我正在寻找一种使用row_number()函数来进行query.delete()的适当方法。
以下是我当前的解决方案：
subquery = session.query(Foo,
func.row_number().over(order_by=desc(Foo.bar)).label("row_number")).subquery()
subquery = session.query(subquery).filter(subquery.c.row_number > 3)
subquery = subquery.from_self(Foo.id)
query = session.query(Foo).filter(
Foo.id.in_(subquery)
)
我不喜欢我需要明确检查Foo.id是否在ID列表中。我需要比较所有对象的列而不是硬编码它们。类似于：
subquery = session.query(Foo,
func.row_number().over(order_by=desc(Foo.bar)).label("row_number")).subquery()
subquery = session.query(subquery).filter((subquery.c.row_number > 3))
subquery = subquery.from_self(Foo)
query = session.query(Foo).filter(
Foo.in_(subquery)
)
但是当然这不起作用。
如果有人能提出解决方案，我将不胜感激。
英文:
I'm looking for an appropriate way to make query.delete() using row_number() function.
Below there is my current solution:
subquery = session.query(Foo,
func.row_number().over(order_by=desc(Foo.bar)).label("row_number")).subquery()
subquery = session.query(subquery).filter(subquery.c.row_number > 3)
subquery = subquery.from_self(Foo.id)
query = session.query(Foo).filter(
Foo.id.in_(subquery)
)
I don't like that I need explicilty check if Foo.id in list of ids. I need to compare all object's columns without hardcoding them. Somthing like:
subquery = session.query(Foo,
func.row_number().over(order_by=desc(Foo.bar)).label("row_number")).subquery()
subquery = session.query(subquery).filter((subquery.c.row_number > 3))
subquery = subquery.from_self(Foo)
query = session.query(Foo).filter(
Foo.in_(subquery)
)
But of course this does not work.
I would be grateful if anyone will suggest a solution.
答案1
得分: 0
经过一些调查，我了解到以下内容：
-
您需要为筛选指定确切的模型字段。我创建了一个名为 _get_reference_attrs 的方法，该方法使用主键字段，如果没有主键，则使用所有字段。
-
有问题的代码使用了3个嵌套的查询（不包括删除查询），但只需要2个。为此，您需要正确收集您在上面选择中选择的内容（attrs_subquery）。
这是我的解决方案：
from sqlalchemy import func, desc, tuple_, inspect
def _get_reference_attrs(model):
"""
收集将在删除查询中使用的模型属性
默认情况下，使用主键列。如果没有主键，则使用所有列。
"""
mapper = inspect(model)
primary_key_names = [primary_key.key for primary_key in mapper.primary_key]
if primary_key_names:
attrs = [attr for attr in mapper.attrs if attr.key in primary_key_names]
else:
attrs = mapper.attrs
return attrs
attrs = _get_reference_attrs(Foo)
subquery = session.query(*attrs,
func.row_number().over(order_by=desc(Foo.bar)).label("row_number")).subquery()
attrs_subquery = [getattr(subquery.c, attr.key) for attr in attrs]
subquery = session.query(*attrs_subquery).select_from(subquery).filter(subquery.c.row_number > 1)
query = session.query(Foo).filter(
tuple_(*attrs).in_(subquery)
)
希望这对您有所帮助。
英文:
After some time of investigations, I understood the following:
- You need to specify exact model fields for filtration. I created a method _get_reference_attrs that uses primary key fields, or all fields, if there is no primary key.
- Code in question used 3 nested select queries (not including delete query), but only 2 are required. For this you need to correctly collect what you are selecting in the upper selection (attrs_subquery)
That's my solution:
from sqlalchemy import func, desc, tuple_, inspect
def _get_reference_attrs(model):
"""
Collecting attributes of model that will be used in delete query
By default the primary key columns are used. If no primary key - all columns are used.
"""
mapper = inspect(model)
primary_key_names = [primary_key.key for primary_key in mapper.primary_key]
if primary_key_names:
attrs = [attr for attr in mapper.attrs if attr.key in primary_key_names]
else:
attrs = mapper.attrs
return attrs
attrs = _get_reference_attrs(Foo)
subquery = session.query(*attrs,
func.row_number().over(order_by=desc(Foo.bar)).label("row_number")).subquery()
attrs_subquery = [getattr(subquery.c, attr.key) for attr in attrs]
subquery = session.query(*attrs_subquery).select_from(subquery).filter(subquery.c.row_number > 1)
query = session.query(Foo).filter(
tuple_(*attrs).in_(subquery)
)
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论