Source code for trackmania.ad

import json
import logging
from contextlib import suppress
from typing import Dict, List

import redis
from typing_extensions import Self

from trackmania.errors import TMIOException

from ._util import _regex_it
from .api import _APIClient
from .config import Client
from .constants import _TMIO

_log = logging.getLogger(__name__)

__all__ = ("Ad",)


async def _get_ad_list() -> List[Dict]:
    ad_list = []

    _log.debug("Getting all ads")
    cache_client = Client._get_cache_client()
    with suppress(ConnectionRefusedError, redis.exceptions.ConnectionError):
        if cache_client.exists("ads"):
            _log.debug("Found all ads in cache")
            ads = json.loads(cache_client.get("ads").decode("utf-8"))
            for ad_dict in ads.get("ads"):
                ad_list.append(ad_dict)
            return ad_list

    api_client = _APIClient()
    all_ads = await api_client.get(_TMIO.build([_TMIO.TABS.ADS]))
    await api_client.close()

    with suppress(KeyError, TypeError):
        raise TMIOException(all_ads["error"])
    with suppress(ConnectionRefusedError, redis.exceptions.ConnectionError):
        _log.debug("Caching all ads for 12hours")
        cache_client.set("ads", json.dumps(all_ads), ex=43200)

    for ad_dict in all_ads.get("ads"):
        ad_list.append(ad_dict)

    return ad_list