34 lines
1.2 KiB
Python
34 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class WikeloProjectsError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class WikeloProjectsClient:
|
|
APP_ID = "695be2905c0b4866dfb21265"
|
|
|
|
def __init__(self, base_url: str = "https://wikelo-projects.com") -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
|
|
async def list_ship_projects(self) -> list[dict[str, Any]]:
|
|
body = await self._get_json(f"{self.base_url}/api/apps/{self.APP_ID}/entities/ShipProject")
|
|
if not isinstance(body, list):
|
|
raise WikeloProjectsError("Wikelo ship projects response was not a list.")
|
|
return [item for item in body if isinstance(item, dict)]
|
|
|
|
async def _get_json(self, url: str) -> Any:
|
|
async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
|
|
response = await client.get(url, headers={"Accept": "application/json"})
|
|
try:
|
|
body = response.json()
|
|
except ValueError as exc:
|
|
raise WikeloProjectsError(f"Wikelo Projects returned non-JSON response: HTTP {response.status_code}") from exc
|
|
if response.status_code >= 400:
|
|
raise WikeloProjectsError(f"Wikelo Projects HTTP {response.status_code}: {body}")
|
|
return body
|