| 1404 | |
| 1405 | |
| 1406 | class ExistsQuery(AwaitableQuery): |
| 1407 | __slots__ = ( |
| 1408 | "_force_indexes", |
| 1409 | "_use_indexes", |
| 1410 | ) |
| 1411 | |
| 1412 | def __init__( |
| 1413 | self, |
| 1414 | model: type[MODEL], |
| 1415 | db: BaseDBAsyncClient, |
| 1416 | q_objects: list[Q], |
| 1417 | annotations: dict[str, Any], |
| 1418 | custom_filters: dict[str, FilterInfoDict], |
| 1419 | force_indexes: set[str], |
| 1420 | use_indexes: set[str], |
| 1421 | ) -> None: |
| 1422 | super().__init__(model) |
| 1423 | self._q_objects = q_objects |
| 1424 | self._db = db |
| 1425 | self._annotations = annotations |
| 1426 | self._custom_filters = custom_filters |
| 1427 | self._force_indexes = force_indexes |
| 1428 | self._use_indexes = use_indexes |
| 1429 | |
| 1430 | def _make_query(self) -> None: |
| 1431 | self.query = copy(self.model._meta.basequery) |
| 1432 | self.resolve_filters() |
| 1433 | self.query._limit = self.query._wrapper_cls(1) |
| 1434 | self.query._select_other(ValueWrapper(1, allow_parametrize=False)) # type:ignore[arg-type] |
| 1435 | |
| 1436 | if self._force_indexes: |
| 1437 | self.query._force_indexes = [] |
| 1438 | self.query = self.query.force_index(*self._force_indexes) |
| 1439 | if self._use_indexes: |
| 1440 | self.query._use_indexes = [] |
| 1441 | self.query = self.query.use_index(*self._use_indexes) |
| 1442 | |
| 1443 | def __await__(self) -> Generator[Any, None, bool]: |
| 1444 | self._choose_db_if_not_chosen() |
| 1445 | self._make_query() |
| 1446 | return self._execute().__await__() |
| 1447 | |
| 1448 | async def _execute( |
| 1449 | self, |
| 1450 | ) -> bool: |
| 1451 | result, _ = await self._db.execute_query(*self.query.get_parameterized_sql()) |
| 1452 | return bool(result) |
| 1453 | |
| 1454 | |
| 1455 | class CountQuery(AwaitableQuery): |
no outgoing calls
no test coverage detected
searching dependent graphs…