ignis-client-python/ignis_client/async_client.py

from __future__ import annotations

from typing import Any

import httpx

from .models import (
    SCENES,
    BlinkResponse,
    CommandRequest,
    DeleteStatusResponse,
    DeviceControlResponse,
    DeviceStatusResponse,
    GroupControlResponse,
    GroupStatusResponse,
    IgnisError,
    KeyActionRequest,
    RescanResponse,
    ScheduleCreateResponse,
    ScheduleCronRequest,
    ScheduleOnceRequest,
    ScheduleTasksResponse,
    ServerInfoResponse,
)


class AsyncIgnisClient:
    def __init__(self, base_url: str, api_key: str, *, timeout: float = 10.0):
        self._base = base_url.rstrip("/")
        self._api_key = api_key
        self._client = httpx.AsyncClient(
            base_url=self._base,
            headers={"X-API-Key": self._api_key},
            timeout=httpx.Timeout(timeout),
            follow_redirects=True,
        )

    async def close(self):
        await self._client.aclose()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self.close()

    async def _request(self, method: str, path: str, **kwargs) -> Any:
        resp = await self._client.request(method, path, **kwargs)
        if resp.status_code >= 400:
            try:
                detail = resp.json().get("detail", resp.text)
            except Exception:
                detail = resp.text or f"HTTP {resp.status_code}"
            raise IgnisError(resp.status_code, detail)
        return resp.json()

    async def _get(self, path: str, **kwargs) -> Any:
        return await self._request("GET", path, **kwargs)

    async def _post(self, path: str, **kwargs) -> Any:
        return await self._request("POST", path, **kwargs)

    async def _delete(self, path: str, **kwargs) -> Any:
        return await self._request("DELETE", path, **kwargs)

    # ------------------------------------------------------------------ Auth

    async def auth_me(self) -> dict[str, Any]:
        return await self._get("/auth/me")

    # ------------------------------------------------------------------ System

    async def system_info(self) -> ServerInfoResponse:
        data = await self._get("/system/info")
        return ServerInfoResponse(**data)

    # ------------------------------------------------------------------ Devices

    async def list_devices(self) -> dict[str, dict[str, str]]:
        return await self._get("/devices")

    async def list_groups(self) -> dict[str, dict[str, Any]]:
        return await self._get("/devices/groups")

    async def list_scenes(self) -> dict[str, int]:
        return dict(SCENES)

    async def create_group(
        self, group_id: str, name: str, macs: list[str]
    ) -> dict[str, Any]:
        return await self._post(
            "/devices/groups", json={"id": group_id, "name": name, "macs": macs}
        )

    async def delete_group(self, group_id: str) -> dict[str, Any]:
        return await self._delete(f"/devices/groups/{group_id}")

    async def rescan(self) -> RescanResponse:
        data = await self._post("/devices/rescan")
        return RescanResponse(**data)

    # ------------------------------------------------------------------ Control

    async def control_device(
        self, device_id: str, payload: CommandRequest
    ) -> DeviceControlResponse:
        data = await self._post(
            f"/control/device/{device_id}",
            json=payload.model_dump(exclude_none=True),
        )
        return DeviceControlResponse(**data)

    async def control_group(
        self, group_id: str, payload: CommandRequest
    ) -> GroupControlResponse:
        data = await self._post(
            f"/control/group/{group_id}",
            json=payload.model_dump(exclude_none=True),
        )
        return GroupControlResponse(**data)

    async def blink_device(self, device_id: str) -> BlinkResponse:
        data = await self._post(f"/control/device/{device_id}/blink")
        return BlinkResponse(**data)

    async def device_status(self, device_id: str) -> DeviceStatusResponse:
        data = await self._get(f"/control/device/{device_id}/status")
        return DeviceStatusResponse(**data)

    async def group_status(self, group_id: str) -> GroupStatusResponse:
        data = await self._get(f"/control/group/{group_id}/status")
        return GroupStatusResponse(**data)

    # --------------------------------------------------------------- Schedules

    async def create_once_schedule(
        self, payload: ScheduleOnceRequest
    ) -> ScheduleCreateResponse:
        data = await self._post(
            "/schedules/once",
            json=payload.model_dump(exclude_none=True),
        )
        return ScheduleCreateResponse(**data)

    async def create_cron_schedule(
        self, payload: ScheduleCronRequest
    ) -> ScheduleCreateResponse:
        data = await self._post(
            "/schedules/cron",
            json=payload.model_dump(exclude_none=True),
        )
        return ScheduleCreateResponse(**data)

    async def list_schedules(self) -> ScheduleTasksResponse:
        data = await self._get("/schedules/tasks")
        return ScheduleTasksResponse(**data)

    async def delete_schedule(self, job_id: str) -> DeleteStatusResponse:
        data = await self._delete(f"/schedules/{job_id}")
        return DeleteStatusResponse(**data)

    # ------------------------------------------------------------------- Stats

    async def stats_summary(self, days: int = 7) -> dict[str, Any]:
        return await self._get("/stats/summary", params={"days": days})

    async def stats_log(self, limit: int = 50) -> list[dict[str, Any]]:
        return await self._get("/stats/log", params={"limit": limit})

    # ---------------------------------------------------------------- API Keys

    async def list_api_keys(self) -> list[dict[str, Any]]:
        return await self._get("/api-keys")

    async def create_api_key(
        self, name: str, *, is_admin: bool = False
    ) -> dict[str, Any]:
        return await self._post(
            "/api-keys", params={"name": name, "is_admin": is_admin}
        )

    async def revoke_api_key(self, key_or_id: str) -> dict[str, Any]:
        return await self._post(
            "/api-keys/revoke", json=KeyActionRequest(key=key_or_id).model_dump()
        )

    async def activate_api_key(self, key_or_id: str) -> dict[str, Any]:
        return await self._post(
            "/api-keys/activate", json=KeyActionRequest(key=key_or_id).model_dump()
        )