Handling error responses
The response.status module provides tools to handle error responses in such a way that future requests
can be made successfully seamlessly.
These objects are designed to be used exclusively with the RequestHandler.
Basic usage
A StatusHandler should not be used directly to handle responses. The aim of a StatusHandler
is to be passed to the RequestHandler to handle responses there.
Each StatusHandler should be built to accept the args and kwargs as per the
RequestHandler method quoted below.
async def _handle_response(self, response: aiohttp.ClientResponse, retry_timer: Timer | None = None) -> bool:
if response.status not in self.response_handlers:
return False
response_handler: StatusHandler = self.response_handlers[response.status]
return await response_handler(
response=response,
authoriser=self.authoriser,
session=self.session,
payload_handler=self.payload_handler,
wait_timer=self.wait_timer,
retry_timer=retry_timer,
)
The status handler will return True for a handled response, and False if it could not handle the response.
It will raise a StatusHandlerError if the status code of the response does not match the status code
of the response.
See also
For more info on how to pass StatusHandler objects to the RequestHandler,
see Handling error responses.
Supported status handlers
The following is a list of each implemented status handler included in this module with excerpts from their
source code showing the supported status codes and the StatusHandler.handle() logic they implement.
See also
You may wish to extend this functionality.
ClientErrorStatusHandler
@property
def status_codes(self) -> list[int]:
return [status.value for status in HTTPStatus if 400 <= status.value < 500]
async def handle(self, response: ClientResponse, *_, **__) -> NoReturn:
self.match(response=response, fail_on_error=True)
self._log(response=response, message="Bad response received and cannot handle or continue processing.")
raise ResponseError(message=await response.text(errors="ignore"), response=response)
RateLimitStatusHandler
@property
def status_codes(self) -> list[int]:
return [429]
async def handle(
self,
response: ClientResponse,
wait_timer: Timer | None = None,
retry_timer: Timer | None = None,
*_,
**__
) -> bool:
self.match(response=response, fail_on_error=True)
if wait_timer is not None:
self._increase_wait(response=response, wait_timer=wait_timer)
if "retry-after" not in response.headers:
return False
wait_seconds = int(response.headers["retry-after"])
wait_dt_str = (datetime.now() + timedelta(seconds=wait_seconds)).strftime("%Y-%m-%d %H:%M:%S")
if retry_timer is not None and wait_seconds > retry_timer.total: # exception if too long
raise ResponseError(
"Rate limit exceeded and wait time is greater than remaining timeout "
f"of {retry_timer.total_remaining:.2f} seconds. Retry again at {wait_dt_str}"
)
if not self._wait_logged:
self.logger.warning(f"\33[93mRate limit exceeded. Retrying again at {wait_dt_str}\33[0m")
self._wait_logged = True
await asyncio.sleep(wait_seconds)
self._wait_logged = False
return True
Writing a StatusHandler
To implement a StatusHandler, you will need to implement the abstract methods as shown below.
class StatusHandler(ABC):
"""
Handles a response that matches the status conditions of this handler
according to its status, payload, headers, etc.
"""
__slots__ = ("logger",)
@property
@abstractmethod
def status_codes(self) -> list[int]:
"""The response status codes this handler can handle."""
raise NotImplementedError
def __init__(self):
#: The :py:class:`logging.Logger` for this object
self.logger: logging.Logger = logging.getLogger(__name__)
def match(self, response: ClientResponse, fail_on_error: bool = False) -> bool:
"""
Check if this handler settings match the given response.
:param response: The response to match.
:param fail_on_error: Raise an exception if the response does not match this handler.
:raise StatusHandlerError: If match fails and `fail_on_error` is True.
"""
match = response.status in self.status_codes
if not match and fail_on_error:
raise StatusHandlerError(
"Response status does not match this handler | "
f"Response={response.status} | Valid={",".join(map(str, self.status_codes))}"
)
return match
def _log(self, response: ClientResponse, message: str = ""):
status = HTTPStatus(response.status)
log = [f"Status: {response.status} ({status.phrase}) - {status.description}"]
if message:
log.append(message)
self.logger.debug(" | ".join(log))
@abstractmethod
async def handle(self, response: ClientResponse, *_, **__) -> bool:
"""
Handle the response based on its status, payload response etc.
:param response: The response to handle.
:return: True if the response was handled, False if it was not.
"""
raise NotImplementedError
def __call__(self, response: ClientResponse, *args, **kwargs) -> Awaitable[bool]:
return self.handle(response, *args, **kwargs)
As an example, the following implements the UnauthorisedStatusHandler.
class UnauthorisedStatusHandler(StatusHandler):
"""Handles unauthorised response status codes by re-authorising credentials through an :py:class:`Authoriser`."""
__slots__ = ()
@property
def status_codes(self) -> list[int]:
return [401]
async def handle(
self,
response: ClientResponse,
authoriser: Authoriser | None = None,
session: ClientSession | None = None,
*_,
**__,
) -> bool:
self.match(response=response, fail_on_error=True)
if authoriser is None or session is None:
return False
self._log(response=response, message="Re-authorising...")
headers = await authoriser
session.headers.update(headers)
return True