qq_lib.core.repeater

 1# Released under MIT License.
 2# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
 3
 4"""
 5Utility for repeated execution with per-item error handling.
 6
 7This module provides the `Repeater` class, which runs a function over a list of
 8items while capturing exceptions, invoking registered handlers, and tracking
 9errors on a per-item basis.
10"""
11
12from collections.abc import Callable
13from typing import Any, Self
14
15
16class Repeater:
17    """
18    Execute a given function repeatedly for a collection of items,
19    with optional per-exception handling and tracking of encountered errors.
20
21    Attributes:
22        items (list[Any]): List of items to process.
23        encountered_errors (dict[int, BaseException]): A dictionary mapping
24            item indices to exceptions encountered during execution.
25        current_iteration (int): The index of the item currently being processed.
26
27    Args:
28        items (list[Any]): A list of items to iterate over.
29        func (Callable): Function to execute for each item. The item will be passed
30            as the first argument, followed by any `*args` and `**kwargs`.
31        *args (Any): Positional arguments forwarded to `func`.
32        **kwargs (Any): Keyword arguments forwarded to `func`.
33    """
34
35    def __init__(
36        self,
37        items: list[Any],
38        func: Callable,
39        *args: Any,
40        **kwargs: Any,
41    ):
42        self.encountered_errors = {}
43        self.items = items
44        self.current_iteration = 0
45
46        self._handlers: dict[
47            type[BaseException], Callable[[BaseException, Self], Any]
48        ] = {}
49        self._func = func
50        self._args = args
51        self._kwargs = kwargs
52
53    def on_exception(
54        self,
55        exc_type: type[BaseException],
56        handler: Callable[[BaseException, Self], Any],
57    ) -> None:
58        """
59        Register a handler function for a specific exception type.
60
61        Args:
62            exc_type (type[BaseException]): The exception type to handle.
63            handler (Callable): Function to call when `exc_type` is raised.
64                The handler must accept two arguments:
65                - BaseException: The caught exception instance.
66                - Repeater: Reference to this `Repeater` instance.
67        """
68        self._handlers[exc_type] = handler
69
70    def run(self) -> None:
71        """
72        Execute the target function for all items, invoking handlers for exceptions.
73
74        Iterates over all items, calling the provided function with each one.
75        If an exception is raised and a handler for that exception type
76        is registered, the handler is called.
77
78        Unhandled exceptions propagate normally and interrupt the iteration.
79
80        Exceptions are recorded in `encountered_errors`, mapping the item's
81        index to the raised exception instance.
82        """
83        for i, item in enumerate(self.items):
84            self.current_iteration = i
85            try:
86                self._func(item, *self._args, **self._kwargs)
87            except tuple(self._handlers.keys()) as e:
88                self.encountered_errors[i] = e
89                handler = self._handlers[type(e)]
90                handler(e, self)