A question about filter queries, is it unsupported or a bug?

[cmdragon]

tortoise-orm == 0.24.0

tags = fields.ManyToManyField(
        'blog.Tags',
        related_name='blogs',
        description='tag',
        through="blog_blog_tags",
        forward_key="tags_id", 
        backward_key="blog_id"  
    )

I have a simple query requirement that needs to retrieve data with tag ids 469 and 7107 simultaneously, but there is a prerequisite condition: only need to query simultaneously when new is true.

Initially, the code was like this

blog_queryset = Blog.filter(tags__id=469)
if new:
    blog_queryset.filter(tags__id=7107)

But I found that the query results are empty，When handling many-to-many relationships, using .filter(tags__id=X).filter(tags__id=Y) consecutively does not produce an intersection query.

Then I tried to query using Q expressions.

query.filter(Q(tags__id=469) & Q(tags__id=7107))

I found that the result is still empty. But strangely, when I changed & to and, I was able to get the correct query results, which were no longer empty.

query.filter(Q(tags__id=469) and Q(tags__id=7107))

[waketzheng]

Usage:

query.filter(Q(tags__id=469) | Q(tags__id=7107))

Or

query.filter(tags__id__in=[469, 7107])

[cmdragon]

Usage:

query.filter(Q(tags__id=469) | Q(tags__id=7107))
Or

query.filter(tags__id__in=[469, 7107])

It needs to be noted that it should include both, not either

[waketzheng]

Example::

from typing import Self

from tortoise import Model, fields, run_async
from tortoise.contrib.test import init_memory_sqlite
from tortoise.expressions import Q, Subquery


class Tags(Model):
    name = fields.CharField(50)


class Blog(Model):
    title = fields.CharField(50)
    tags = fields.ManyToManyField(
        "models.Tags",
        related_name="blogs",
        description="tag",
        through="blog_blog_tags",
        forward_key="tags_id",
        backward_key="blog_id",
    )

    @classmethod
    async def filter_by_tags(cls, id1: int, id2: int | None = None) -> list[Self]:
        query = cls.filter(tags__id=id1)
        if id2:
            query = query.filter(
                pk__in=Subquery(
                    Blog.filter(tags__id=id2).values_list(cls._meta.pk_attr, flat=True)
                )
            )
        return await query


@init_memory_sqlite
async def main() -> None:
    t1 = await Tags.create(name="1")
    t2 = await Tags.create(name="2")
    b1 = await Blog.create(title="1")
    b2 = await Blog.create(title="2")
    b3 = await Blog.create(title="3")
    await b1.tags.add(t1)
    await b2.tags.add(t2)
    await b3.tags.add(t1, t2)

    assert (await Blog.filter(tags__id=t1.pk)) == [b1, b3]
    assert (await Blog.filter(tags__id=t2.pk)) == [b2, b3]
    new = True
    q = Q(tags__id=t1.pk)
    if new:
        q &= Q(
            pk__in=Subquery(Blog.filter(tags__id=t2.pk).values_list("id", flat=True))
        )
    query = Blog.filter(q)
    # print(query.sql())
    assert (await query) == [b3]
    both = await Blog.filter_by_tags(t1.pk, t2.pk)
    print(f"{both = }")
    assert both == [b3]


if __name__ == "__main__":
    run_async(main())

[cmdragon]

Example::

from typing import Self

from tortoise import Model, fields, run_async
from tortoise.contrib.test import init_memory_sqlite
from tortoise.expressions import Q, Subquery

class Tags(Model):
name = fields.CharField(50)

class Blog(Model):
title = fields.CharField(50)
tags = fields.ManyToManyField(
"models.Tags",
related_name="blogs",
description="tag",
through="blog_blog_tags",
forward_key="tags_id",
backward_key="blog_id",
)

@classmethod
async def filter_by_tags(cls, id1: int, id2: int | None = None) -> list[Self]:
    query = cls.filter(tags__id=id1)
    if id2:
        query = query.filter(
            pk__in=Subquery(
                Blog.filter(tags__id=id2).values_list(cls._meta.pk_attr, flat=True)
            )
        )
    return await query

@init_memory_sqlite
async def main() -> None:
t1 = await Tags.create(name="1")
t2 = await Tags.create(name="2")
b1 = await Blog.create(title="1")
b2 = await Blog.create(title="2")
b3 = await Blog.create(title="3")
await b1.tags.add(t1)
await b2.tags.add(t2)
await b3.tags.add(t1, t2)

assert (await Blog.filter(tags__id=t1.pk)) == [b1, b3]
assert (await Blog.filter(tags__id=t2.pk)) == [b2, b3]
new = True
q = Q(tags__id=t1.pk)
if new:
    q &= Q(
        pk__in=Subquery(Blog.filter(tags__id=t2.pk).values_list("id", flat=True))
    )
query = Blog.filter(q)
# print(query.sql())
assert (await query) == [b3]
both = await Blog.filter_by_tags(t1.pk, t2.pk)
print(f"{both = }")
assert both == [b3]

if name == "main":
run_async(main())

First, thank you for your positive response, but it is not necessary.
query.filter(Q(tags__id=469) and Q(tags__id=7107)) This code can solve the problem.

I want to ask is. When handling many-to-many relationships, using .filter(tags__id=X).filter(tags__id=Y) consecutively does not produce an intersection query. Is it supported to use it as smoothly as using Django ORM?

[waketzheng]

I want to ask is. When handling many-to-many relationships, using .filter(tags__id=X).filter(tags__id=Y) consecutively does not produce an intersection query. Is it supported to use it as smoothly as using Django ORM?

Currently, not. But PR is welcome.

Please note that in my example:

qs = await Blog.filter(tags__id=t1.pk, pk__in=Subquery(Blog.filter(tags__id=t2.pk).values_list("id", flat=True))
print(qs)
# Got: [b3]
qs2 = await Blog.filter(Q(tags__id=t1.pk) and Q(tags__id=t2.pk))
print(qs2)
# Got: [b2, b3]

That is to say and may not act as you want.