async-retry-fetch.py

"""General utilities"""
import asyncio
import json
import socket
import time

from tornado.httpclient import AsyncHTTPClient, HTTPClientError
from tornado.log import app_log
from tornado.simple_httpclient import HTTPTimeoutError


async def fetch(url_or_req, *args, timeout=20, **kwargs):
    """Fetch with a wrapper to retry and log errors"""

    try:
        url = url_or_req.url
        method = url_or_req.method
    except AttributeError:
        url = url_or_req
        method = kwargs.get("method", "GET")

    log_url = f"{method} {url.split('?', 1)[0]}"
    deadline = time.perf_counter() + timeout

    async def retry_connections():
        app_log.info(f"{log_url}")
        try:
            return await AsyncHTTPClient().fetch(url_or_req, *args, **kwargs)
        except (TimeoutError, HTTPTimeoutError, socket.gaierror) as e:
            app_log.error(f"Socket error fetching {log_url}: {e}")
            return False
        except HTTPClientError as e:
            # retry on server availability errors
            if e.code in {502, 503, 599}:
                app_log.error(f"Error fetching {log_url}: {e}")
                return False
            elif e.code == 429:
                retry_after_header = e.response.headers.get("Retry-After")
                try:
                    retry_after = int(retry_after_header)
                except Exception as e:
                    app_log.error(f"Failed to handle Retry-After: {retry_after_header}")
                    retry_after = 30
                app_log.error(
                    f"Rate limit fetching {log_url}: {e} retrying after {retry_after}s"
                )
                max_sleep = max(0, deadline - time.perf_counter())
                await asyncio.sleep(min(max_sleep, retry_after))
                return False
            else:
                raise

    try:
        return await exponential_backoff(
            retry_connections, timeout=timeout, fail_message=""
        )
    except HTTPClientError as e:
        if e.response is None:
            app_log.error(f"Error fetching {log_url}: {e}")
            raise

        if e.response and e.response.body:
            message = e.response.body.decode("utf-8", "replace")
            try:
                body_json = json.loads(message)
                message = body_json["error_description"]
            except (KeyError, ValueError):
                pass
            app_log.error(f"Error fetching {log_url}: {message[:1024]}")
        raise