Add API-Reference
495
API-Reference.-.md
Normal file
495
API-Reference.-.md
Normal file
@@ -0,0 +1,495 @@
|
||||
# 🛠️ API Reference
|
||||
|
||||
Technical documentation for developers working with AdGuard Control Hub. This covers the internal API, entity structure, and integration architecture.
|
||||
|
||||
## 📡 AdGuard Home API Integration
|
||||
|
||||
### Supported API Endpoints
|
||||
|
||||
The integration interacts with these AdGuard Home API endpoints:
|
||||
|
||||
| Endpoint | Method | Purpose |
|
||||
|----------|--------|---------|
|
||||
| `/control/status` | GET | Server status and configuration |
|
||||
| `/control/clients` | GET | List all configured clients |
|
||||
| `/control/clients/add` | POST | Add new client |
|
||||
| `/control/clients/update` | POST | Update existing client |
|
||||
| `/control/clients/delete` | POST | Remove client |
|
||||
| `/control/stats` | GET | DNS query statistics |
|
||||
| `/control/protection` | POST | Enable/disable protection |
|
||||
| `/control/blocked_services/all` | GET | Available blocked services |
|
||||
| `/control/blocked_services/get` | GET | Client blocked services |
|
||||
| `/control/blocked_services/update` | POST | Update blocked services |
|
||||
|
||||
### API Authentication
|
||||
|
||||
```python
|
||||
# Basic authentication used for all API calls
|
||||
from aiohttp import BasicAuth
|
||||
|
||||
auth = BasicAuth(username, password)
|
||||
headers = {"Content-Type": "application/json"}
|
||||
```
|
||||
|
||||
### Error Handling
|
||||
|
||||
```python
|
||||
# API wrapper error handling pattern
|
||||
async def _request(self, method: str, endpoint: str, data: dict = None) -> dict:
|
||||
try:
|
||||
async with self.session.request(method, url, json=data, headers=headers, auth=auth) as response:
|
||||
response.raise_for_status()
|
||||
if response.status == 204 or not response.content_length:
|
||||
return {}
|
||||
return await response.json()
|
||||
except aiohttp.ClientError as err:
|
||||
_LOGGER.error("API request failed: %s", err)
|
||||
raise
|
||||
except asyncio.TimeoutError:
|
||||
_LOGGER.error("API request timeout")
|
||||
raise
|
||||
```
|
||||
|
||||
## 🏗️ Integration Architecture
|
||||
|
||||
### Component Structure
|
||||
|
||||
```
|
||||
adguard_hub/
|
||||
├── __init__.py # Main integration setup
|
||||
├── api.py # AdGuard Home API wrapper
|
||||
├── config_flow.py # Configuration UI flow
|
||||
├── const.py # Constants and configuration
|
||||
├── switch.py # Switch platform implementation
|
||||
├── sensor.py # Sensor platform (optional)
|
||||
├── binary_sensor.py # Binary sensor platform (optional)
|
||||
├── services.py # Custom services
|
||||
└── strings.json # UI strings and translations
|
||||
```
|
||||
|
||||
### Data Flow
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
A[Home Assistant] --> B[AdGuard Control Hub]
|
||||
B --> C[Update Coordinator]
|
||||
C --> D[AdGuard Home API]
|
||||
D --> E[AdGuard Home Server]
|
||||
|
||||
C --> F[Switch Platform]
|
||||
C --> G[Sensor Platform]
|
||||
C --> H[Binary Sensor Platform]
|
||||
|
||||
F --> I[Entity Registry]
|
||||
G --> I
|
||||
H --> I
|
||||
```
|
||||
|
||||
### Update Coordinator Pattern
|
||||
|
||||
```python
|
||||
class AdGuardControlHubCoordinator(DataUpdateCoordinator):
|
||||
"""Manage data fetching from AdGuard Home."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, api: AdGuardHomeAPI):
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=f"{DOMAIN}_coordinator",
|
||||
update_interval=timedelta(seconds=SCAN_INTERVAL),
|
||||
)
|
||||
self.api = api
|
||||
|
||||
async def _async_update_data(self):
|
||||
"""Fetch data from AdGuard Home."""
|
||||
try:
|
||||
# Concurrent API calls for better performance
|
||||
results = await asyncio.gather(
|
||||
self.api.get_clients(),
|
||||
self.api.get_statistics(),
|
||||
self.api.get_status(),
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
# Process and return structured data
|
||||
return self._process_api_results(results)
|
||||
except Exception as err:
|
||||
raise UpdateFailed(f"Error fetching data: {err}")
|
||||
```
|
||||
|
||||
## 🔌 Entity System
|
||||
|
||||
### Entity Types and Hierarchy
|
||||
|
||||
```python
|
||||
# Base entity class
|
||||
class AdGuardBaseEntity(CoordinatorEntity):
|
||||
"""Base class for AdGuard entities."""
|
||||
|
||||
def __init__(self, coordinator: AdGuardControlHubCoordinator, api: AdGuardHomeAPI):
|
||||
super().__init__(coordinator)
|
||||
self.api = api
|
||||
self._attr_device_info = self._create_device_info()
|
||||
|
||||
def _create_device_info(self) -> dict:
|
||||
"""Create device info for entity grouping."""
|
||||
return {
|
||||
"identifiers": {(DOMAIN, f"{self.api.host}:{self.api.port}")},
|
||||
"name": f"AdGuard Control Hub ({self.api.host})",
|
||||
"manufacturer": MANUFACTURER,
|
||||
"model": "AdGuard Home",
|
||||
"sw_version": self.coordinator.protection_status.get("version"),
|
||||
}
|
||||
```
|
||||
|
||||
### Switch Entities
|
||||
|
||||
```python
|
||||
# Global protection switch
|
||||
class AdGuardProtectionSwitch(AdGuardBaseEntity, SwitchEntity):
|
||||
"""Control global AdGuard protection."""
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
return f"{self.api.host}_{self.api.port}_protection"
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
return self.coordinator.protection_status.get("protection_enabled", False)
|
||||
|
||||
async def async_turn_on(self, **kwargs):
|
||||
await self.api.set_protection(True)
|
||||
await self.coordinator.async_request_refresh()
|
||||
|
||||
# Client-specific switches
|
||||
class AdGuardClientSwitch(AdGuardBaseEntity, SwitchEntity):
|
||||
"""Control per-client protection."""
|
||||
|
||||
def __init__(self, coordinator, api, client_name: str):
|
||||
super().__init__(coordinator, api)
|
||||
self.client_name = client_name
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
return f"{self.api.host}_{self.api.port}_client_{self.client_name}"
|
||||
```
|
||||
|
||||
### Sensor Entities
|
||||
|
||||
```python
|
||||
class AdGuardStatsSensor(AdGuardBaseEntity, SensorEntity):
|
||||
"""DNS statistics sensor."""
|
||||
|
||||
@property
|
||||
def native_value(self) -> int:
|
||||
return self.coordinator.statistics.get(self.stat_key, 0)
|
||||
|
||||
@property
|
||||
def device_class(self) -> str:
|
||||
return SensorDeviceClass.TIMESTAMP if "time" in self.stat_key else None
|
||||
```
|
||||
|
||||
## 🔧 Custom Services
|
||||
|
||||
### Service Registration
|
||||
|
||||
```python
|
||||
async def async_register_services(hass: HomeAssistant, api: AdGuardHomeAPI) -> None:
|
||||
"""Register integration services."""
|
||||
|
||||
async def emergency_unblock_service(call):
|
||||
"""Handle emergency unblock service."""
|
||||
duration = call.data.get("duration", 300)
|
||||
clients = call.data.get("clients", ["all"])
|
||||
|
||||
try:
|
||||
if "all" in clients:
|
||||
await api.set_protection(False)
|
||||
# Schedule re-enabling after duration
|
||||
hass.async_create_task(_restore_protection_after_delay(duration))
|
||||
else:
|
||||
# Handle specific clients
|
||||
for client_name in clients:
|
||||
await api.temporary_disable_client(client_name, duration)
|
||||
except Exception as err:
|
||||
_LOGGER.error("Emergency unblock failed: %s", err)
|
||||
raise HomeAssistantError(f"Service call failed: {err}")
|
||||
|
||||
# Register service with schema validation
|
||||
hass.services.async_register(
|
||||
DOMAIN,
|
||||
"emergency_unblock",
|
||||
emergency_unblock_service,
|
||||
schema=EMERGENCY_UNBLOCK_SCHEMA,
|
||||
)
|
||||
```
|
||||
|
||||
### Service Schemas
|
||||
|
||||
```python
|
||||
import voluptuous as vol
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
|
||||
# Service parameter validation schemas
|
||||
EMERGENCY_UNBLOCK_SCHEMA = vol.Schema({
|
||||
vol.Optional("duration", default=300): cv.positive_int,
|
||||
vol.Optional("clients", default=["all"]): vol.All(cv.ensure_list, [cv.string]),
|
||||
})
|
||||
|
||||
BLOCK_SERVICES_SCHEMA = vol.Schema({
|
||||
vol.Required("client_name"): cv.string,
|
||||
vol.Required("services"): vol.All(cv.ensure_list, [cv.string]),
|
||||
vol.Optional("schedule"): vol.Schema({
|
||||
vol.Required("enabled"): cv.boolean,
|
||||
vol.Optional("time_zone", default="Local"): cv.string,
|
||||
vol.Optional("mon"): TIME_RANGE_SCHEMA,
|
||||
vol.Optional("tue"): TIME_RANGE_SCHEMA,
|
||||
# ... other days
|
||||
}),
|
||||
})
|
||||
|
||||
TIME_RANGE_SCHEMA = vol.Schema({
|
||||
vol.Required("start"): cv.time,
|
||||
vol.Required("end"): cv.time,
|
||||
})
|
||||
```
|
||||
|
||||
## 📊 Data Models
|
||||
|
||||
### Client Data Structure
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class AdGuardClient:
|
||||
"""Represent an AdGuard Home client."""
|
||||
name: str
|
||||
ids: list[str]
|
||||
use_global_settings: bool = True
|
||||
use_global_blocked_services: bool = True
|
||||
blocked_services: list[str] = field(default_factory=list)
|
||||
filtering_enabled: bool = True
|
||||
parental_enabled: bool = False
|
||||
safebrowsing_enabled: bool = False
|
||||
safesearch_enabled: bool = False
|
||||
upstream_dns: list[str] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def from_api_response(cls, data: dict) -> "AdGuardClient":
|
||||
"""Create client from API response."""
|
||||
return cls(
|
||||
name=data.get("name", ""),
|
||||
ids=data.get("ids", []),
|
||||
use_global_settings=data.get("use_global_settings", True),
|
||||
# ... map other fields
|
||||
)
|
||||
|
||||
def to_api_request(self) -> dict:
|
||||
"""Convert to API request format."""
|
||||
return {
|
||||
"name": self.name,
|
||||
"ids": self.ids,
|
||||
"use_global_settings": self.use_global_settings,
|
||||
# ... map all fields
|
||||
}
|
||||
```
|
||||
|
||||
### Statistics Data Structure
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class AdGuardStatistics:
|
||||
"""AdGuard Home DNS statistics."""
|
||||
dns_queries: int
|
||||
blocked_filtering: int
|
||||
replaced_safebrowsing: int
|
||||
replaced_safesearch: int
|
||||
replaced_parental: int
|
||||
avg_processing_time: float
|
||||
|
||||
@property
|
||||
def blocked_percentage(self) -> float:
|
||||
"""Calculate blocking percentage."""
|
||||
if self.dns_queries == 0:
|
||||
return 0.0
|
||||
return (self.blocked_filtering / self.dns_queries) * 100
|
||||
```
|
||||
|
||||
## 🔄 Configuration Flow
|
||||
|
||||
### Configuration Steps
|
||||
|
||||
```python
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle configuration flow."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle user input."""
|
||||
errors = {}
|
||||
|
||||
if user_input is not None:
|
||||
try:
|
||||
# Validate connection
|
||||
info = await self._validate_input(user_input)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
else:
|
||||
# Create config entry
|
||||
return self.async_create_entry(
|
||||
title=info["title"],
|
||||
data=user_input
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=self._get_schema(user_input),
|
||||
errors=errors
|
||||
)
|
||||
|
||||
async def _validate_input(self, data: dict) -> dict:
|
||||
"""Validate user input and test connection."""
|
||||
api = AdGuardHomeAPI(
|
||||
host=data[CONF_HOST],
|
||||
port=data[CONF_PORT],
|
||||
username=data.get(CONF_USERNAME),
|
||||
password=data.get(CONF_PASSWORD),
|
||||
ssl=data.get(CONF_SSL, False),
|
||||
session=async_get_clientsession(self.hass),
|
||||
)
|
||||
|
||||
# Test connection
|
||||
if not await api.test_connection():
|
||||
raise CannotConnect
|
||||
|
||||
# Get server info for title
|
||||
status = await api.get_status()
|
||||
return {"title": f"AdGuard Control Hub ({data[CONF_HOST]})"}
|
||||
```
|
||||
|
||||
## 🧪 Testing Framework
|
||||
|
||||
### Unit Test Structure
|
||||
|
||||
```python
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
from homeassistant.core import HomeAssistant
|
||||
from custom_components.adguard_hub.api import AdGuardHomeAPI
|
||||
|
||||
@pytest.fixture
|
||||
def mock_api():
|
||||
"""Mock API for testing."""
|
||||
api = MagicMock(spec=AdGuardHomeAPI)
|
||||
api.get_status = AsyncMock(return_value={"protection_enabled": True})
|
||||
api.get_clients = AsyncMock(return_value={"clients": []})
|
||||
return api
|
||||
|
||||
async def test_protection_switch_on(hass: HomeAssistant, mock_api):
|
||||
"""Test protection switch turns on correctly."""
|
||||
coordinator = AdGuardControlHubCoordinator(hass, mock_api)
|
||||
switch = AdGuardProtectionSwitch(coordinator, mock_api)
|
||||
|
||||
await switch.async_turn_on()
|
||||
|
||||
mock_api.set_protection.assert_called_once_with(True)
|
||||
coordinator.async_request_refresh.assert_called_once()
|
||||
```
|
||||
|
||||
### Integration Testing
|
||||
|
||||
```python
|
||||
async def test_integration_setup(hass: HomeAssistant):
|
||||
"""Test integration setup."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
CONF_HOST: "192.168.1.100",
|
||||
CONF_PORT: 3000,
|
||||
CONF_USERNAME: "admin",
|
||||
CONF_PASSWORD: "password",
|
||||
},
|
||||
)
|
||||
|
||||
with patch("custom_components.adguard_hub.api.AdGuardHomeAPI.test_connection", return_value=True):
|
||||
config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
|
||||
assert config_entry.state == config_entries.ConfigEntryState.LOADED
|
||||
```
|
||||
|
||||
## 🔍 Debugging and Logging
|
||||
|
||||
### Logging Configuration
|
||||
|
||||
```python
|
||||
import logging
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
# Log levels for different scenarios
|
||||
_LOGGER.debug("Detailed debugging information")
|
||||
_LOGGER.info("General information")
|
||||
_LOGGER.warning("Warning about potential issues")
|
||||
_LOGGER.error("Error occurred but recoverable")
|
||||
_LOGGER.exception("Exception occurred with traceback")
|
||||
```
|
||||
|
||||
### Performance Monitoring
|
||||
|
||||
```python
|
||||
import time
|
||||
from functools import wraps
|
||||
|
||||
def monitor_performance(func):
|
||||
"""Decorator to monitor function performance."""
|
||||
@wraps(func)
|
||||
async def wrapper(*args, **kwargs):
|
||||
start_time = time.time()
|
||||
try:
|
||||
result = await func(*args, **kwargs)
|
||||
duration = time.time() - start_time
|
||||
_LOGGER.debug("%s completed in %.2fs", func.__name__, duration)
|
||||
return result
|
||||
except Exception as err:
|
||||
duration = time.time() - start_time
|
||||
_LOGGER.error("%s failed after %.2fs: %s", func.__name__, duration, err)
|
||||
raise
|
||||
return wrapper
|
||||
```
|
||||
|
||||
## 📈 Metrics and Monitoring
|
||||
|
||||
### Integration Health Metrics
|
||||
|
||||
```python
|
||||
class HealthMetrics:
|
||||
"""Track integration health metrics."""
|
||||
|
||||
def __init__(self):
|
||||
self.api_calls_total = 0
|
||||
self.api_calls_failed = 0
|
||||
self.last_successful_update = None
|
||||
self.last_error = None
|
||||
|
||||
def record_api_call(self, success: bool, error: Exception = None):
|
||||
"""Record API call metrics."""
|
||||
self.api_calls_total += 1
|
||||
if success:
|
||||
self.last_successful_update = datetime.now()
|
||||
else:
|
||||
self.api_calls_failed += 1
|
||||
self.last_error = str(error) if error else "Unknown error"
|
||||
|
||||
@property
|
||||
def success_rate(self) -> float:
|
||||
"""Calculate API call success rate."""
|
||||
if self.api_calls_total == 0:
|
||||
return 0.0
|
||||
return ((self.api_calls_total - self.api_calls_failed) / self.api_calls_total) * 100
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
**For more technical details**, see the source code in the repository or the [Development Guide](Development) for setting up a development environment.
|
||||
Reference in New Issue
Block a user