|
|
|
"""The Gitea integration."""
|
|
|
|
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
|
|
from homeassistant.const import CONF_ACCESS_TOKEN, Platform
|
|
|
|
from homeassistant.core import HomeAssistant, callback
|
|
|
|
from homeassistant.helpers import device_registry as dr
|
|
|
|
from homeassistant.helpers.aiohttp_client import (
|
|
|
|
SERVER_SOFTWARE,
|
|
|
|
async_get_clientsession,
|
|
|
|
)
|
|
|
|
|
|
|
|
import giteapy
|
|
|
|
from giteapy import RepositoryApi
|
|
|
|
|
|
|
|
from .const import CONF_REPOSITORIES, DOMAIN, LOGGER
|
|
|
|
from .coordinator import GiteaDataUpdateCoordinator
|
|
|
|
|
|
|
|
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
|
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|
|
|
"""Set up GitHub from a config entry."""
|
|
|
|
hass.data.setdefault(DOMAIN, {})
|
|
|
|
|
|
|
|
configuration = giteapy.Configuration()
|
|
|
|
configuration.api_key['access_token'] = entry.data[CONF_ACCESS_TOKEN]
|
|
|
|
|
|
|
|
client = giteapy.RepositoryApi(giteapy.ApiClient(configuration))
|
|
|
|
|
|
|
|
repositories: list[str] = entry.options[CONF_REPOSITORIES]
|
|
|
|
|
|
|
|
for repository in repositories:
|
|
|
|
coordinator = GiteaDataUpdateCoordinator(
|
|
|
|
hass=hass,
|
|
|
|
client=client,
|
|
|
|
repository=repository,
|
|
|
|
)
|
|
|
|
|
|
|
|
await coordinator.async_config_entry_first_refresh()
|
|
|
|
|
|
|
|
hass.data[DOMAIN][repository] = coordinator
|
|
|
|
|
|
|
|
async_cleanup_device_registry(hass=hass, entry=entry)
|
|
|
|
|
|
|
|
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
|
|
|
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
|
|
|
|
return True
|
|
|
|
|
|
|
|
@callback
|
|
|
|
def async_cleanup_device_registry(
|
|
|
|
hass: HomeAssistant,
|
|
|
|
entry: ConfigEntry,
|
|
|
|
) -> None:
|
|
|
|
"""Remove entries form device registry if we no longer track the repository."""
|
|
|
|
device_registry = dr.async_get(hass)
|
|
|
|
devices = dr.async_entries_for_config_entry(
|
|
|
|
registry=device_registry,
|
|
|
|
config_entry_id=entry.entry_id,
|
|
|
|
)
|
|
|
|
for device in devices:
|
|
|
|
for item in device.identifiers:
|
|
|
|
if DOMAIN == item[0] and item[1] not in entry.options[CONF_REPOSITORIES]:
|
|
|
|
LOGGER.debug(
|
|
|
|
"Unlinking device %s for untracked repository %s from config entry %s",
|
|
|
|
device.id,
|
|
|
|
item[1],
|
|
|
|
entry.entry_id,
|
|
|
|
)
|
|
|
|
device_registry.async_update_device(
|
|
|
|
device.id, remove_config_entry_id=entry.entry_id
|
|
|
|
)
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|
|
|
"""Unload a config entry."""
|
|
|
|
repositories: dict[str, GiteaDataUpdateCoordinator] = hass.data[DOMAIN]
|
|
|
|
for coordinator in repositories.values():
|
|
|
|
coordinator.unsubscribe()
|
|
|
|
|
|
|
|
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
|
|
|
hass.data.pop(DOMAIN)
|
|
|
|
return unload_ok
|
|
|
|
|
|
|
|
|
|
|
|
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
|
|
|
|
"""Handle an options update."""
|
|
|
|
await hass.config_entries.async_reload(entry.entry_id)
|