Initial commit

Signed-off-by: Rafal Zielinski <sq4ind@gmail.com>

custom_components/adguard_control_hub/api.py

@@ -0,0 +1,154 @@
+"""AdGuard Home API client."""
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+import aiohttp
+import async_timeout
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import (
+    API_CLIENTS,
+    API_DNS_CONFIG,
+    API_STATUS,
+    API_STATS,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class AdGuardHomeApiError(Exception):
+    """Exception to indicate a general API error."""
+
+
+class AdGuardHomeConnectionError(AdGuardHomeApiError):
+    """Exception to indicate a connection error."""
+
+
+class AdGuardHomeAuthError(AdGuardHomeApiError):
+    """Exception to indicate an authentication error."""
+
+
+class AdGuardHomeAPI:
+    """AdGuard Home API client."""
+
+    def __init__(
+        self,
+        session: aiohttp.ClientSession,
+        host: str,
+        port: int,
+        username: str | None = None,
+        password: str | None = None,
+        ssl: bool = False,
+        verify_ssl: bool = True,
+    ) -> None:
+        """Initialize the API client."""
+        self._session = session
+        self._host = host
+        self._port = port
+        self._username = username
+        self._password = password
+        self._ssl = ssl
+        self._verify_ssl = verify_ssl
+
+        protocol = "https" if ssl else "http"
+        self._base_url = f"{protocol}://{host}:{port}"
+
+    async def _request(
+        self,
+        method: str,
+        endpoint: str,
+        data: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Make a request to the AdGuard Home API."""
+        url = f"{self._base_url}{endpoint}"
+        auth = None
+
+        if self._username and self._password:
+            auth = aiohttp.BasicAuth(self._username, self._password)
+
+        headers = {"Content-Type": "application/json"}
+
+        try:
+            async with async_timeout.timeout(10):
+                async with self._session.request(
+                    method,
+                    url,
+                    json=data,
+                    auth=auth,
+                    headers=headers,
+                    ssl=self._verify_ssl,
+                ) as response:
+                    if response.status == 401:
+                        raise AdGuardHomeAuthError("Authentication failed")
+
+                    if response.status == 403:
+                        raise AdGuardHomeAuthError("Access forbidden")
+
+                    if response.status not in (200, 204):
+                        text = await response.text()
+                        raise AdGuardHomeApiError(
+                            f"Request failed with status {response.status}: {text}"
+                        )
+
+                    if response.status == 204:
+                        return {}
+
+                    return await response.json()
+
+        except asyncio.TimeoutError as err:
+            raise AdGuardHomeConnectionError("Timeout connecting to AdGuard Home") from err
+        except aiohttp.ClientError as err:
+            raise AdGuardHomeConnectionError(f"Error connecting to AdGuard Home: {err}") from err
+
+    async def get_status(self) -> dict[str, Any]:
+        """Get AdGuard Home status."""
+        return await self._request("GET", API_STATUS)
+
+    async def get_stats(self) -> dict[str, Any]:
+        """Get AdGuard Home statistics."""
+        return await self._request("GET", API_STATS)
+
+    async def get_clients(self) -> dict[str, Any]:
+        """Get AdGuard Home clients."""
+        return await self._request("GET", API_CLIENTS)
+
+    async def set_protection(self, enabled: bool) -> None:
+        """Enable or disable protection."""
+        data = {"protection_enabled": enabled}
+        await self._request("POST", API_DNS_CONFIG, data)
+
+    async def set_filtering(self, enabled: bool) -> None:
+        """Enable or disable filtering."""
+        data = {"filtering_enabled": enabled}
+        await self._request("POST", API_DNS_CONFIG, data)
+
+    async def set_safebrowsing(self, enabled: bool) -> None:
+        """Enable or disable safe browsing."""
+        data = {"safebrowsing_enabled": enabled}
+        await self._request("POST", API_DNS_CONFIG, data)
+
+    async def set_parental_control(self, enabled: bool) -> None:
+        """Enable or disable parental control."""
+        data = {"parental_enabled": enabled}
+        await self._request("POST", API_DNS_CONFIG, data)
+
+    async def set_safe_search(self, enabled: bool) -> None:
+        """Enable or disable safe search."""
+        data = {"safesearch_enabled": enabled}
+        await self._request("POST", API_DNS_CONFIG, data)
+
+    async def set_query_log(self, enabled: bool) -> None:
+        """Enable or disable query log."""
+        data = {"querylog_enabled": enabled}
+        await self._request("POST", API_DNS_CONFIG, data)
+
+    async def test_connection(self) -> bool:
+        """Test connection to AdGuard Home."""
+        try:
+            await self.get_status()
+            return True
+        except AdGuardHomeApiError:
+            return False