Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions ormar/queryset/queries/filter_query.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,14 +30,14 @@ def apply(
:rtype: sqlalchemy.sql.selectable.Select
"""
if self.filter_clauses:
if len(self.filter_clauses) == 1:
clause: Union[TextClause, ColumnElement[Any]] = self.filter_clauses[
0
].get_text_clause()
else:
clause = sqlalchemy.sql.and_(
*[x.get_text_clause() for x in self.filter_clauses]
)
clause = sqlalchemy.sql.not_(clause) if self.exclude else clause
clauses: list[Union[TextClause, ColumnElement[Any]]] = [
x.get_text_clause() for x in self.filter_clauses
]
if self.exclude:
# every exclude() call is its own NOT, all of them are ANDed
clauses = [sqlalchemy.sql.not_(x) for x in clauses]
clause: Union[TextClause, ColumnElement[Any]] = (
clauses[0] if len(clauses) == 1 else sqlalchemy.sql.and_(*clauses)
)
expr = expr.where(clause)
return expr
19 changes: 13 additions & 6 deletions ormar/queryset/queryset.py
Original file line number Diff line number Diff line change
Expand Up @@ -337,14 +337,21 @@ def filter( # noqa: A003
select_related=select_related,
filter_clauses=self.filter_clauses,
)
if _exclude:
own_clauses, select_related = qryclause.prepare_filter(
_own_only=True, **kwargs
)
if not own_clauses and not filter_groups:
return self.rebuild_self(select_related=select_related)
exclude_group = FilterGroup(*filter_groups)
exclude_group.actions = own_clauses
return self.rebuild_self(
exclude_clauses=self.exclude_clauses + [exclude_group],
select_related=select_related,
)
filter_clauses, select_related = qryclause.prepare_filter(**kwargs)
filter_clauses = filter_clauses + filter_groups # type: ignore
if _exclude:
exclude_clauses = filter_clauses
filter_clauses = self.filter_clauses
else:
exclude_clauses = self.exclude_clauses
filter_clauses = filter_clauses
exclude_clauses = self.exclude_clauses

return self.rebuild_self(
filter_clauses=filter_clauses,
Expand Down
92 changes: 92 additions & 0 deletions tests/test_queries/test_chained_exclude.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
import pytest

import ormar
from ormar.exceptions import QueryDefinitionError
from tests.lifespan import init_tests
from tests.settings import create_config

base_ormar_config = create_config()


class Item(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="items")

id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=50)
kind: str = ormar.String(max_length=50)


create_test_database = init_tests(base_ormar_config)


async def create_items():
for name, kind in [("a", "x"), ("b", "y"), ("c", "z"), ("d", "x")]:
await Item.objects.create(name=name, kind=kind)


@pytest.mark.asyncio
async def test_chained_excludes_all_apply():
async with base_ormar_config.database:
async with base_ormar_config.database.transaction(force_rollback=True):
await create_items()

qs = Item.objects.exclude(kind="x").exclude(name="b")
items = await qs.order_by("id").all()
assert [item.name for item in items] == ["c"]

items = await Item.objects.filter(kind="x").exclude(name="a").all()
assert [item.name for item in items] == ["d"]

items = await Item.objects.exclude(kind="x", name="a").order_by("id").all()
assert [item.name for item in items] == ["b", "c", "d"]


@pytest.mark.asyncio
async def test_chained_excludes_update_and_delete():
async with base_ormar_config.database:
async with base_ormar_config.database.transaction(force_rollback=True):
await create_items()

qs = Item.objects.exclude(kind="x").exclude(name="b")
assert await qs.update(kind="ZZ") == 1
assert await Item.objects.filter(kind="ZZ").count() == 1

assert await Item.objects.exclude(kind="ZZ").exclude(name="b").delete() == 2
assert await Item.objects.count() == 2


@pytest.mark.asyncio
async def test_empty_exclude_is_noop():
async with base_ormar_config.database:
async with base_ormar_config.database.transaction(force_rollback=True):
await create_items()

assert await Item.objects.exclude().count() == 4
assert await Item.objects.exclude(**{}).count() == 4
with pytest.raises(QueryDefinitionError):
await Item.objects.exclude().delete()


@pytest.mark.asyncio
async def test_chained_exclude_with_groups_and_limit():
async with base_ormar_config.database:
async with base_ormar_config.database.transaction(force_rollback=True):
await create_items()

items = (
await Item.objects.exclude(ormar.or_(kind="x", name="b"))
.exclude(name="z")
.order_by("id")
.all()
)
assert [item.name for item in items] == ["c"]

items = (
await Item.objects.exclude(name="a")
.exclude(name="b")
.order_by("id")
.limit(1)
.offset(1)
.all()
)
assert [item.name for item in items] == ["d"]
Loading