Query used to fetch related models in subsequent queries. Each model is fetched only ones by the name of the relation. That means that for each prefetch_related entry next query is issued to database.
| 465 | |
| 466 | |
| 467 | class PrefetchQuery: |
| 468 | """ |
| 469 | Query used to fetch related models in subsequent queries. |
| 470 | Each model is fetched only ones by the name of the relation. |
| 471 | That means that for each prefetch_related entry next query is issued to database. |
| 472 | """ |
| 473 | |
| 474 | def __init__( # noqa: CFQ002 |
| 475 | self, |
| 476 | model_cls: type["Model"], |
| 477 | excludable: "ExcludableItems", |
| 478 | prefetch_related: list, |
| 479 | select_related: list, |
| 480 | orders_by: list["OrderAction"], |
| 481 | ) -> None: |
| 482 | self.model = model_cls |
| 483 | self.excludable = excludable |
| 484 | self.select_dict = translate_list_to_dict(select_related, default={}) |
| 485 | self.prefetch_dict = translate_list_to_dict(prefetch_related, default={}) |
| 486 | self.orders_by = orders_by |
| 487 | self.load_tasks: list[Node] = [] |
| 488 | |
| 489 | async def prefetch_related(self, models: Sequence["Model"]) -> Sequence["Model"]: |
| 490 | """ |
| 491 | Main entry point for prefetch_query. |
| 492 | |
| 493 | Receives list of already initialized parent models with all children from |
| 494 | select_related already populated. Receives also list of row sql result rows |
| 495 | as it's quicker to extract ids that way instead of calling each model. |
| 496 | |
| 497 | Returns list with related models already prefetched and set. |
| 498 | |
| 499 | :param models: list of already instantiated models from main query |
| 500 | :type models: Sequence[Model] |
| 501 | :param rows: row sql result of the main query before the prefetch |
| 502 | :type rows: list[sqlalchemy.engine.result.RowProxy] |
| 503 | :return: list of models with children prefetched |
| 504 | :rtype: list[Model] |
| 505 | """ |
| 506 | parent_task = RootNode(models=cast(list["Model"], models)) |
| 507 | self._build_load_tree( |
| 508 | prefetch_dict=self.prefetch_dict, |
| 509 | select_dict=self.select_dict, |
| 510 | parent=parent_task, |
| 511 | model=self.model, |
| 512 | ) |
| 513 | await parent_task.load_data() |
| 514 | parent_task.reload_tree() |
| 515 | return parent_task.models |
| 516 | |
| 517 | def _build_load_tree( |
| 518 | self, |
| 519 | select_dict: dict, |
| 520 | prefetch_dict: dict, |
| 521 | parent: Node, |
| 522 | model: type["Model"], |
| 523 | ) -> None: |
| 524 | """ |
no outgoing calls
no test coverage detected