diff --git a/CHANGELOG.rst b/CHANGELOG.rst index b111c9525..bae4a8d4a 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -17,6 +17,7 @@ Added - ``QuerySet.contains()`` method to check if an object exists in a queryset. - Added comprehensive EXPLAIN support for MySQL and PostgreSQL. - Built-in ``DomainNameValidator``, ``URLValidator``, and ``EmailValidator`` classes for common validation patterns. (#2162) +- `AwaitableQuery.as_query()` was added back (#2194) Fixed ^^^^^ diff --git a/tests/test_queryset.py b/tests/test_queryset.py index b59238b1a..bd768a913 100644 --- a/tests/test_queryset.py +++ b/tests/test_queryset.py @@ -1140,3 +1140,42 @@ async def test_union_with_annotate_raises(db): with pytest.raises(ParamsError, match="Union queries do not support annotations"): await qs1.union(qs2) + + +@pytest.mark.asyncio +async def test_as_query_queryset(db): + sql = IntFields.all().as_query().get_sql() + assert sql.startswith("SELECT") + + +@pytest.mark.asyncio +async def test_as_query_update(db): + sql = IntFields.filter(intnum=10).update(intnum=99).as_query().get_sql() + assert sql.startswith("UPDATE") + + +@pytest.mark.asyncio +async def test_as_query_delete(db): + sql = IntFields.filter(intnum=10).delete().as_query().get_sql() + assert sql.startswith("DELETE") + + +@pytest.mark.asyncio +async def test_as_query_exists(db): + sql = IntFields.filter(intnum=10).exists().as_query().get_sql() + assert "SELECT 1" in sql + + +@pytest.mark.asyncio +async def test_as_query_count(db): + sql = IntFields.filter(intnum=10).count().as_query().get_sql() + assert "COUNT(*)" in sql + + +@pytest.mark.asyncio +async def test_as_query_union_count(db): + qs1 = IntFields.filter(intnum__gte=50).only("id", "intnum") + qs2 = IntFields.filter(intnum__lt=50).only("id", "intnum") + sql = qs1.union(qs2).count().as_query().get_sql() + assert "UNION" in sql + assert "COUNT(*)" in sql diff --git a/tortoise/queryset.py b/tortoise/queryset.py index aecffc42d..1f26a2cb6 100644 --- a/tortoise/queryset.py +++ b/tortoise/queryset.py @@ -316,6 +316,12 @@ def sql(self, params_inline=False) -> str: sql, _ = self.query.get_parameterized_sql() return sql + def as_query(self) -> QueryBuilder: + """Return the internal pypika query object.""" + self._choose_db_if_not_chosen() + self._make_query() + return self.query + def _make_query(self) -> None: raise NotImplementedError() # pragma: nocoverage