(self, poolers_by_task: Mapping[PoolingTask, Pooler])
| 527 | """Dispatches calls to a sub-pooler based on the pooling task.""" |
| 528 | |
| 529 | def __init__(self, poolers_by_task: Mapping[PoolingTask, Pooler]) -> None: |
| 530 | super().__init__() |
| 531 | |
| 532 | for task, pooler in poolers_by_task.items(): |
| 533 | if task not in pooler.get_supported_tasks(): |
| 534 | raise ValueError( |
| 535 | f"{pooler=} does not support {task=}. " f"Supported tasks: {pooler.get_supported_tasks()}" |
| 536 | ) |
| 537 | |
| 538 | self.poolers_by_task = poolers_by_task |
| 539 | |
| 540 | def get_supported_tasks(self) -> Set[PoolingTask]: |
| 541 | return set(self.poolers_by_task) |
nothing calls this directly
no test coverage detected