| 1453 | |
| 1454 | |
| 1455 | class CountQuery(AwaitableQuery): |
| 1456 | __slots__ = ( |
| 1457 | "_limit", |
| 1458 | "_offset", |
| 1459 | "_force_indexes", |
| 1460 | "_use_indexes", |
| 1461 | ) |
| 1462 | |
| 1463 | def __init__( |
| 1464 | self, |
| 1465 | model: type[MODEL], |
| 1466 | db: BaseDBAsyncClient, |
| 1467 | q_objects: list[Q], |
| 1468 | annotations: dict[str, Any], |
| 1469 | custom_filters: dict[str, FilterInfoDict], |
| 1470 | limit: int | None, |
| 1471 | offset: int | None, |
| 1472 | force_indexes: set[str], |
| 1473 | use_indexes: set[str], |
| 1474 | ) -> None: |
| 1475 | super().__init__(model) |
| 1476 | self._q_objects = q_objects |
| 1477 | self._annotations = annotations |
| 1478 | self._custom_filters = custom_filters |
| 1479 | self._limit = limit |
| 1480 | self._offset = offset or 0 |
| 1481 | self._db = db |
| 1482 | self._force_indexes = force_indexes |
| 1483 | self._use_indexes = use_indexes |
| 1484 | |
| 1485 | def _make_query(self) -> None: |
| 1486 | self.query = copy(self.model._meta.basequery) |
| 1487 | self.resolve_filters() |
| 1488 | count_term = Count(Star()) |
| 1489 | if self.query._groupbys: |
| 1490 | count_term = count_term.over() |
| 1491 | |
| 1492 | # remove annotations |
| 1493 | self.query._selects = [] |
| 1494 | self.query._select_other(count_term) |
| 1495 | |
| 1496 | if self._force_indexes: |
| 1497 | self.query._force_indexes = [] |
| 1498 | self.query = self.query.force_index(*self._force_indexes) |
| 1499 | if self._use_indexes: |
| 1500 | self.query._use_indexes = [] |
| 1501 | self.query = self.query.use_index(*self._use_indexes) |
| 1502 | |
| 1503 | def __await__(self) -> Generator[Any, None, int]: |
| 1504 | self._choose_db_if_not_chosen() |
| 1505 | self._make_query() |
| 1506 | return self._execute().__await__() |
| 1507 | |
| 1508 | async def _execute(self) -> int: |
| 1509 | _, result = await self._db.execute_query(*self.query.get_parameterized_sql()) |
| 1510 | if not result: |
| 1511 | return 0 |
| 1512 | count = list(dict(result[0]).values())[0] - self._offset |
no outgoing calls
no test coverage detected
searching dependent graphs…