51 lines
2.0 KiB
Python
51 lines
2.0 KiB
Python
from typing import Any
|
|
import requests
|
|
from . import __version__
|
|
from .config import BASE_URL
|
|
|
|
class APIRequestError(Exception):
|
|
"""Raised when the API request is malformed"""
|
|
|
|
class APIResponseError(Exception):
|
|
"""Raised when the API returns an error"""
|
|
|
|
class APIClient:
|
|
def __init__(self) -> None:
|
|
self.base_url = BASE_URL.rstrip("/")
|
|
self.session = requests.Session()
|
|
self.session.headers.update({"User-Agent": f"kaylee's berghain-cli/{__version__}"})
|
|
|
|
def new_game(self, player_id: str | None, scenario: int) -> dict[str, Any]:
|
|
"""
|
|
Start a new game for this player
|
|
"""
|
|
if player_id is None:
|
|
raise APIRequestError("A player ID is required")
|
|
return self.__do_request("new-game",{"scenario": scenario, "playerId": player_id}, "gameId")
|
|
|
|
def decide_and_next(self, game_id: str, person_index: int, accept: bool | None = None) -> dict[str, Any]:
|
|
"""Submit a decision for the current person and get the next one"""
|
|
params: dict[str, Any] = {"gameId": game_id, "personIndex": person_index}
|
|
|
|
if person_index > 0 and accept is None:
|
|
raise APIRequestError("You must specify whether to accept or reject this person")
|
|
|
|
if accept is not None:
|
|
params["accept"] = str(accept).lower()
|
|
|
|
return self.__do_request("decide-and-next", params, "status")
|
|
|
|
|
|
def __do_request(self, endpoint: str, params: dict[str, Any], check_for: str | None = None) -> dict[str, Any]:
|
|
resp = self.session.get(f"{self.base_url}/{endpoint}", params=params, timeout=10)
|
|
if not resp.ok:
|
|
raise APIResponseError(f"HTTP {resp.status_code}: {resp.text}")
|
|
try:
|
|
data = resp.json()
|
|
except ValueError as e:
|
|
raise APIResponseError(f"Invalid JSON in response: {resp.text}")
|
|
if check_for is not None and check_for not in data:
|
|
raise APIResponseError(f"Malformed response (key \"{check_for}\" not found): {data}")
|
|
|
|
return data
|