lpdb_python.async_session

 13class AsyncLpdbSession(AbstractLpdbSession):
 14    """
 15    Asynchronous implementation of a LPDB session
 16    """
 17
 18    __session: aiohttp.ClientSession
 19
 20    def __init__(self, api_key, base_url=AbstractLpdbSession.BASE_URL):
 21        super().__init__(api_key, base_url=base_url)
 22        self.__session = aiohttp.ClientSession(
 23            self._base_url, headers=self._get_header()
 24        )
 25
 26    def __enter__(self) -> None:
 27        raise TypeError("Use async with instead")
 28
 29    def __exit__(
 30        self,
 31        exc_type: Optional[type[BaseException]],
 32        exc_val: Optional[BaseException],
 33        exc_tb: Optional[TracebackType],
 34    ) -> None:
 35        pass
 36
 37    async def __aenter__(self) -> "AsyncLpdbSession":
 38        return self
 39
 40    async def __aexit__(
 41        self,
 42        exc_type: Optional[type[BaseException]],
 43        exc_val: Optional[BaseException],
 44        exc_tb: Optional[TracebackType],
 45    ) -> None:
 46        await self.__session.close()
 47
 48    @staticmethod
 49    async def get_wikis() -> set[str]:
 50        async with aiohttp.ClientSession("https://liquipedia.net/") as session:
 51            async with session.get(
 52                "api.php",
 53                headers={"accept": "application/json", "accept-encoding": "gzip"},
 54            ) as response:
 55                wikis = await response.json()
 56                return set(wikis["allwikis"].keys())
 57
 58    @staticmethod
 59    async def __handle_response(
 60        response: aiohttp.ClientResponse,
 61    ) -> list[dict[str, Any]]:
 62        return AbstractLpdbSession._parse_results(
 63            response.status, await response.json()
 64        )
 65
 66    @override
 67    async def make_request(
 68        self,
 69        lpdb_datatype: LpdbDataType,
 70        wiki: str | list[str],
 71        limit: int = 20,
 72        offset: int = 0,
 73        conditions: Optional[str] = None,
 74        query: Optional[str | list[str]] = None,
 75        order: Optional[str | list[tuple[str, Literal["asc", "desc"]]]] = None,
 76        groupby: Optional[str | list[tuple[str, Literal["asc", "desc"]]]] = None,
 77        **kwargs,
 78    ) -> list[dict[str, Any]]:
 79        if not AbstractLpdbSession._validate_datatype_name(lpdb_datatype):
 80            raise ValueError(f'Invalid LPDB data type: "{lpdb_datatype}"')
 81        async with self.__session.get(
 82            lpdb_datatype,
 83            params=AbstractLpdbSession._parse_params(
 84                wiki=wiki,
 85                limit=limit,
 86                offset=offset,
 87                conditions=conditions,
 88                query=query,
 89                order=order,
 90                groupby=groupby,
 91                **kwargs,
 92            ),
 93        ) as response:
 94            return await AsyncLpdbSession.__handle_response(response)
 95
 96    @override
 97    async def make_count_request(
 98        self,
 99        lpdb_datatype,
100        wiki: str,
101        conditions: Optional[str] = None,
102    ) -> int:
103        response = await self.make_request(
104            lpdb_datatype, wiki=wiki, conditions=conditions, query="count::objectname"
105        )
106        return response[0]["count_objectname"]
107
108    @override
109    async def get_team_template(
110        self, wiki: str, template: str, date: Optional[date] = None
111    ) -> Optional[dict[str, Any]]:
112        params = {
113            "wiki": wiki,
114            "template": template,
115        }
116        if date is not None:
117            params["date"] = date.isoformat()
118        async with self.__session.get("teamtemplate", params=params) as response:
119            parsed_response = await AsyncLpdbSession.__handle_response(response)
120            if parsed_response[0] is None:
121                return None
122            return parsed_response[0]
123
124    @override
125    async def get_team_template_list(
126        self, wiki: str, pagination: int = 1
127    ) -> list[dict[str, Any]]:
128        async with self.__session.get(
129            "teamtemplatelist",
130            params={"wiki": wiki, "pagination": pagination},
131        ) as response:
132            return await AsyncLpdbSession.__handle_response(response)