Skip to content

mirascope.integrations.tenacity

collect_errors

Collects specified errors into an errors keyword argument.

Example:

from mirascope.integrations.tenacity import collect_errors
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3), after=collect_errors(ValueError))
def throw_value_error(*, errors: list[ValueError] | None = None):
    if errors:
        print(errors[-1])
    raise ValueError("Throwing Error")


try:
    throw_value_error()
    # > Throwing Error
    # > Throwing Error
except RetryError:
    ...

Parameters:

Name Type Description Default
*args type[Exception]

The Exception types to catch.

()
Source code in mirascope/retries/tenacity.py
def collect_errors(
    *args: type[Exception],
) -> Callable[[RetryCallState], None]:
    """Collects specified errors into an `errors` keyword argument.

    Example:

    ```python
    from mirascope.integrations.tenacity import collect_errors
    from tenacity import retry, stop_after_attempt

    @retry(stop=stop_after_attempt(3), after=collect_errors(ValueError))
    def throw_value_error(*, errors: list[ValueError] | None = None):
        if errors:
            print(errors[-1])
        raise ValueError("Throwing Error")


    try:
        throw_value_error()
        # > Throwing Error
        # > Throwing Error
    except RetryError:
        ...
    ```

    Args:
        *args: The `Exception` types to catch.
    """

    def inner(retry_state: RetryCallState) -> None:
        """Collect errors into a `errors` kwarg of the wrapped function."""
        if (
            (outcome := retry_state.outcome)
            and (exception := outcome.exception())
            and type(exception) in args
        ):
            errors = retry_state.kwargs.pop("errors", [])
            retry_state.kwargs["errors"] = (
                errors + [exception] if errors else [exception]
            )

    return inner