lpdb_python.session

  1from abc import abstractmethod, ABC
  2from datetime import date
  3from functools import cache
  4from http import HTTPStatus
  5from typing import (
  6    Any,
  7    Final,
  8    Literal,
  9    NotRequired,
 10    Optional,
 11    override,
 12    Required,
 13    TypedDict,
 14    TypeGuard,
 15)
 16import re
 17import warnings
 18import importlib.metadata as metadata
 19
 20import requests
 21
 22__all__ = ["LpdbDataType", "LpdbError", "LpdbWarning", "LpdbSession"]
 23
 24_PACKAGE_NAME: Final[str] = "lpdb_python"
 25
 26
 27@cache
 28def _get_version() -> str:
 29    try:
 30        return metadata.version(_PACKAGE_NAME)
 31    except metadata.PackageNotFoundError:
 32        return "dev"
 33
 34
 35type LpdbDataType = Literal[
 36    "broadcasters",
 37    "company",
 38    "datapoint",
 39    "externalmedialink",
 40    "match",
 41    "placement",
 42    "player",
 43    "series",
 44    "squadplayer",
 45    "standingsentry",
 46    "standingstable",
 47    "team",
 48    "tournament",
 49    "transfer",
 50]
 51"""
 52Python type representing the available data types in LPDB
 53"""
 54
 55
 56class LpdbResponse(TypedDict):
 57    """
 58    Typed representation of a proper LPDB response.
 59    """
 60
 61    result: Required[list[dict[str, Any]]]
 62    """
 63    The result of the query
 64    """
 65    error: NotRequired[list[str]]
 66    """
 67    Errors raised by LPDB
 68    """
 69    warning: NotRequired[list[str]]
 70    """
 71    Non-fatal issues with the LPDB request
 72    """
 73
 74
 75class LpdbError(Exception):
 76    """
 77    Raised when the LPDB request created a fatal issue.
 78    """
 79
 80    pass
 81
 82
 83class LpdbRateLimitError(LpdbError):
 84    """
 85    Raised when the LPDB request failed due to a rate limit.
 86    """
 87
 88    def __init__(self, wiki: str, table: str, *args):
 89        super().__init__(f'Rate limit reached for table "{table}" in "{wiki}"')
 90        self.wiki = wiki
 91        self.table = table
 92
 93
 94class LpdbWarning(Warning):
 95    """
 96    Warnings about LPDB response.
 97    """
 98
 99    pass
100
101
102class AbstractLpdbSession(ABC):
103    """
104    An abstract LPDB session
105    """
106
107    BASE_URL: Final[str] = "https://api.liquipedia.net/api/v3/"
108
109    __DATA_TYPES: Final[frozenset[str]] = frozenset(
110        {
111            "broadcasters",
112            "company",
113            "datapoint",
114            "externalmedialink",
115            "match",
116            "placement",
117            "player",
118            "series",
119            "squadplayer",
120            "standingsentry",
121            "standingstable",
122            "team",
123            "tournament",
124            "transfer",
125        }
126    )
127
128    __api_key: str
129
130    def __init__(self, api_key: str, base_url: str = BASE_URL):
131        self.__api_key = re.sub(r"^ApiKey ", "", api_key)
132        self._base_url = base_url
133
134    @cache
135    def _get_header(self) -> dict[str, str]:
136        return {
137            "authorization": f"Apikey {self.__api_key}",
138            "accept": "application/json",
139            "accept-encoding": "gzip",
140            "user-agent": f"{_PACKAGE_NAME}/{_get_version()}",
141        }
142
143    @staticmethod
144    def _validate_datatype_name(lpdb_datatype: str) -> TypeGuard[LpdbDataType]:
145        return lpdb_datatype in AbstractLpdbSession.__DATA_TYPES
146
147    @staticmethod
148    @abstractmethod
149    def get_wikis() -> set[str]:
150        """
151        Fetches the set of all available wikis.
152
153        :return: set of all available wiki names
154        """
155        pass
156
157    @abstractmethod
158    def make_request(
159        self,
160        lpdb_datatype: LpdbDataType,
161        wiki: str | list[str],
162        limit: int = 20,
163        offset: int = 0,
164        conditions: Optional[str] = None,
165        query: Optional[str | list[str]] = None,
166        order: Optional[str | list[tuple[str, Literal["asc", "desc"]]]] = None,
167        groupby: Optional[str | list[tuple[str, Literal["asc", "desc"]]]] = None,
168        **kwargs,
169    ) -> list[dict[str, Any]]:
170        """
171        Creates an LPDB query request.
172
173        :param lpdb_datatype: the data type to query
174        :param wiki: the wiki(s) to query
175        :param limit: the amount of results wanted
176        :param offset: the offset, the first `offset` results from the query will be dropped
177        :param conditions: the conditions for the query
178        :paran query: the data field(s) to fetch from query
179        :param order: the order of results to be sorted in; each ordering rule can specified as a `(datapoint, direction)` tuple
180        :param groupby: the way that the query results are grouped; each grouping rule can specified as a `(datapoint, direction)` tuple
181
182        :return: result of the query
183
184        :raises ValueError: if an invalid `lpdb_datatype` is supplied
185        :raises LpdbError: if something went wrong with the request
186        """
187        pass
188
189    def make_count_request(
190        self,
191        lpdb_datatype: LpdbDataType,
192        wiki: str,
193        conditions: Optional[str] = None,
194    ) -> int:
195        """
196        Queries the number of objects that satisfy the specified condition(s).
197
198        :param lpdb_datatype: the data type to query
199        :param wiki: the wiki to query
200        :param conditions: the conditions for the query
201
202        :return: number of objects that satisfy the condition(s)
203
204        :raises ValueError: if an invalid `lpdb_datatype` is supplied
205        :raises LpdbError: if something went wrong with the request
206        """
207        pass
208
209    @abstractmethod
210    def get_team_template(
211        self,
212        wiki: str,
213        template: str,
214        date: Optional[date] = None,
215    ) -> Optional[dict[str, Any]]:
216        """
217        Queries a team template from LPDB.
218
219        :param wiki: the wiki to query
220        :param template: the name of team template
221        :param date: the contextual date for the requested team template
222
223        :return: the requested team template, may return `None` if the requested team template does not exist
224
225        :raises LpdbError: if something went wrong with the request
226        """
227        pass
228
229    @abstractmethod
230    def get_team_template_list(
231        self,
232        wiki: str,
233        pagination: int = 1,
234    ) -> list[dict[str, Any]]:
235        """
236        Queries a list of team template from LPDB.
237
238        :param wiki: the wiki to query
239        :param pagination: used for pagination
240
241        :return: team templates
242
243        :raises LpdbError: if something went wrong with the request
244        """
245        pass
246
247    @staticmethod
248    def _parse_params(
249        wiki: str | list[str],
250        limit: int = 20,
251        offset: int = 0,
252        conditions: Optional[str] = None,
253        query: Optional[str | list[str]] = None,
254        order: Optional[str | list[tuple[str, Literal["asc", "desc"]]]] = None,
255        groupby: Optional[str | list[tuple[str, Literal["asc", "desc"]]]] = None,
256        **kwargs,
257    ):
258        parameters = dict(kwargs)
259        if isinstance(wiki, str):
260            parameters["wiki"] = wiki
261        elif isinstance(wiki, list):
262            parameters["wiki"] = ", ".join(wiki)
263        else:
264            raise TypeError()
265        parameters["limit"] = min(limit, 1000)
266        parameters["offset"] = offset
267        if conditions is not None:
268            parameters["conditions"] = conditions
269        if query is not None:
270            if isinstance(query, str):
271                parameters["query"] = query
272            else:
273                parameters["query"] = ", ".join(query)
274        if order is not None:
275            if isinstance(order, str):
276                parameters["order"] = order
277            else:
278                parameters["order"] = ", ".join(
279                    [f"{order_tuple[0]} {order_tuple[1]}" for order_tuple in order]
280                )
281        if groupby is not None:
282            if isinstance(groupby, str):
283                parameters["groupby"] = groupby
284            else:
285                parameters["groupby"] = ", ".join(
286                    [
287                        f"{groupby_tuple[0]} {groupby_tuple[1]}"
288                        for groupby_tuple in groupby
289                    ]
290                )
291        return parameters
292
293    @staticmethod
294    def _parse_results(
295        status_code: int, response: LpdbResponse
296    ) -> list[dict[str, Any]]:
297        result = response["result"]
298        lpdb_warnings = response.get("warning")
299        lpdb_errors = response.get("error")
300
301        if lpdb_errors and len(lpdb_errors) != 0:
302            rate_limit = re.match(
303                r"API key \"[0-9A-Za-z]+\" limits for wiki \"(?P<wiki>[a-z]+)\" and table \"(?P<table>[a-z]+)\" exceeded\.",
304                lpdb_errors[0],
305            )
306            if rate_limit:
307                raise LpdbRateLimitError(
308                    wiki=rate_limit.group("wiki"), table=rate_limit.group("table")
309                )
310            raise LpdbError(re.sub(r"^Error: ?", "", lpdb_errors[0]))
311        elif status_code != HTTPStatus.OK:
312            status = HTTPStatus(status_code)
313            raise LpdbError(f"HTTP {status_code}: {status.name}")
314        if lpdb_warnings and len(lpdb_warnings) != 0:
315            for lpdb_warning in lpdb_warnings:
316                warnings.warn(lpdb_warning, LpdbWarning)
317        return result
318
319
320class LpdbSession(AbstractLpdbSession):
321    """
322    Implementation of a LPDB session
323    """
324
325    @staticmethod
326    def get_wikis() -> set[str]:
327        response = requests.get(
328            "https://liquipedia.net/api.php",
329            headers={"accept": "application/json", "accept-encoding": "gzip"},
330        )
331        wikis = response.json()
332        return set(wikis["allwikis"].keys())
333
334    @staticmethod
335    def __handle_response(response: requests.Response) -> list[dict[str, Any]]:
336        status = HTTPStatus(response.status_code)
337        return AbstractLpdbSession._parse_results(status, response.json())
338
339    @override
340    def make_request(
341        self,
342        lpdb_datatype: LpdbDataType,
343        wiki: str | list[str],
344        limit: int = 20,
345        offset: int = 0,
346        conditions: Optional[str] = None,
347        query: Optional[str | list[str]] = None,
348        order: Optional[str | list[tuple[str, Literal["asc", "desc"]]]] = None,
349        groupby: Optional[str | list[tuple[str, Literal["asc", "desc"]]]] = None,
350        **kwargs,
351    ) -> list[dict[str, Any]]:
352        if not AbstractLpdbSession._validate_datatype_name(lpdb_datatype):
353            raise ValueError(f'Invalid LPDB data type: "{lpdb_datatype}"')
354        lpdb_response = requests.get(
355            self._base_url + lpdb_datatype,
356            headers=self._get_header(),
357            params=AbstractLpdbSession._parse_params(
358                wiki=wiki,
359                limit=limit,
360                offset=offset,
361                conditions=conditions,
362                query=query,
363                order=order,
364                groupby=groupby,
365                **kwargs,
366            ),
367        )
368        return LpdbSession.__handle_response(lpdb_response)
369
370    @override
371    def make_count_request(
372        self,
373        lpdb_datatype: LpdbDataType,
374        wiki: str,
375        conditions: Optional[str] = None,
376    ) -> int:
377        response = self.make_request(
378            lpdb_datatype, wiki, conditions=conditions, query="count::objectname"
379        )
380        return response[0]["count_objectname"]
381
382    @override
383    def get_team_template(
384        self, wiki: str, template: str, date: Optional[date] = None
385    ) -> Optional[dict[str, Any]]:
386        params = {
387            "wiki": wiki,
388            "template": template,
389        }
390        if date is not None:
391            params["date"] = date.isoformat()
392        lpdb_response = requests.get(
393            self._base_url + "teamtemplate",
394            headers=self._get_header(),
395            params=params,
396        )
397        return LpdbSession.__handle_response(lpdb_response)[0]
398
399    @override
400    def get_team_template_list(
401        self, wiki: str, pagination: int = 1
402    ) -> list[dict[str, Any]]:
403        lpdb_response = requests.get(
404            self._base_url + "teamtemplatelist",
405            headers=self._get_header(),
406            params={"wiki": wiki, "pagination": pagination},
407        )
408        return LpdbSession.__handle_response(lpdb_response)