Class AttemptRecorder

tests/test_middleware.py:196–224

Records the `retries_taken` and response status of every attempt the chain runs for.

class AttemptRecorder(Middleware):
    """Records the `retries_taken` and response status of every attempt the chain runs for."""

    def __init__(self) -> None:
        self.attempts: list[int] = []
        self.statuses: list[int] = []
        self.errors: list[Exception] = []

    @override
    def handle(self, request: APIRequest, call_next: CallNext) -> Any:
        self.attempts.append(request.retries_taken)
        try:
            response = call_next(request)
        except Exception as err:
            self.errors.append(err)
            raise
        self.statuses.append(response.status_code)
        return response

    @override
    async def handle_async(self, request: APIRequest, call_next: AsyncCallNext) -> Any:
        self.attempts.append(request.retries_taken)
        try:
            response = await call_next(request)
        except Exception as err:
            self.errors.append(err)
            raise
        self.statuses.append(response.status_code)
        return response


class Boom(Exception):