40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
|
from typing import Any
|
||
|
|
||
|
import aiohttp
|
||
|
|
||
|
ARKHAMDB_ORIGIN = 'https://arkhamdb.adamgoldsmith.name'
|
||
|
|
||
|
ArkhamDBDeck = Any # TODO: better typing
|
||
|
|
||
|
|
||
|
class ArkhamDBClient:
|
||
|
session: aiohttp.ClientSession
|
||
|
|
||
|
origin: str
|
||
|
|
||
|
def __init__(self, arkhamdb_origin: str = ARKHAMDB_ORIGIN):
|
||
|
self.session = aiohttp.ClientSession()
|
||
|
self.origin = arkhamdb_origin
|
||
|
|
||
|
async def get_latest_deck(self, deck_id: int) -> ArkhamDBDeck:
|
||
|
next_deck_id = deck_id
|
||
|
deck = None
|
||
|
while deck is None or deck["next_deck"] is not None:
|
||
|
async with self.session.get(self.origin + f"/api/public/deck/{next_deck_id}.json") as resp:
|
||
|
deck = await resp.json()
|
||
|
next_deck_id = deck["next_deck"]
|
||
|
|
||
|
return deck
|
||
|
|
||
|
async def get_latest_decks(self, deck_ids: dict[int, str]) -> dict[int, tuple[str, ArkhamDBDeck]]:
|
||
|
latest_decks: dict[int, tuple[str, ArkhamDBDeck]] = {}
|
||
|
for deck_id, prefix in deck_ids.items():
|
||
|
try:
|
||
|
deck = await self.get_latest_deck(deck_id)
|
||
|
latest_decks[deck['id']] = (prefix, deck)
|
||
|
except aiohttp.ContentTypeError:
|
||
|
# TODO: json was invalid, should probably alert user
|
||
|
pass
|
||
|
|
||
|
return latest_decks
|