mirror of
https://github.com/hacs/integration.git
synced 2025-07-29 12:50:01 +00:00
45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
"""HACS Data client."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from aiohttp import ClientSession, ClientTimeout
|
|
|
|
from .exceptions import HacsException
|
|
|
|
|
|
class HacsDataClient:
|
|
"""HACS Data client."""
|
|
|
|
def __init__(self, session: ClientSession, client_name: str) -> None:
|
|
"""Initialize."""
|
|
self._client_name = client_name
|
|
self._session = session
|
|
|
|
async def _do_request(
|
|
self,
|
|
filename: str,
|
|
section: str | None = None,
|
|
) -> dict[str, dict[str, Any]] | list[str]:
|
|
"""Do request."""
|
|
endpoint = "/".join([v for v in [section, filename] if v is not None])
|
|
try:
|
|
response = await self._session.get(
|
|
f"https://data-v2.hacs.xyz/{endpoint}",
|
|
timeout=ClientTimeout(total=60),
|
|
headers={"User-Agent": self._client_name},
|
|
)
|
|
response.raise_for_status()
|
|
except Exception as exception:
|
|
raise HacsException(f"Error fetching data from HACS: {exception}") from exception
|
|
|
|
return await response.json()
|
|
|
|
async def get_data(self, section: str | None) -> dict[str, dict[str, Any]]:
|
|
"""Get data."""
|
|
return await self._do_request(filename="data.json", section=section)
|
|
|
|
async def get_repositories(self, section: str) -> list[str]:
|
|
"""Get repositories."""
|
|
return await self._do_request(filename="repositories.json", section=section)
|