coordinator.py

Go to the documentation of this file.

"""Custom data update coordinator for the GitHub integration."""
 
from __future__ import annotations
 
from typing import Any
 
from aiogithubapi import (
    GitHubAPI,
    GitHubConnectionException,
    GitHubEventModel,
    GitHubException,
    GitHubRatelimitException,
    GitHubResponseModel,
)
 
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import FALLBACK_UPDATE_INTERVAL, LOGGER, REFRESH_EVENT_TYPES
 
GRAPHQL_REPOSITORY_QUERY = """
query ($owner: String!, $repository: String!) {
  rateLimit {
    cost
    remaining
  }
  repository(owner: $owner, name: $repository) {
    default_branch_ref: defaultBranchRef {
      commit: target {
        ... on Commit {
          message: messageHeadline
          url
          sha: oid
        }
      }
    }
    stargazers_count: stargazerCount
    forks_count: forkCount
    full_name: nameWithOwner
    id: databaseId
    watchers(first: 1) {
      total: totalCount
    }
    discussion: discussions(
      first: 1
      orderBy: {field: CREATED_AT, direction: DESC}
    ) {
      total: totalCount
      discussions: nodes {
        title
        url
        number
      }
    }
    issue: issues(
      first: 1
      states: OPEN
      orderBy: {field: CREATED_AT, direction: DESC}
    ) {
      total: totalCount
      issues: nodes {
        title
        url
        number
      }
    }
    pull_request: pullRequests(
      first: 1
      states: OPEN
      orderBy: {field: CREATED_AT, direction: DESC}
    ) {
      total: totalCount
      pull_requests: nodes {
        title
        url
        number
      }
    }
    release: latestRelease {
      name
      url
      tag: tagName
    }
    refs(
      first: 1
      refPrefix: "refs/tags/"
      orderBy: {field: TAG_COMMIT_DATE, direction: DESC}
    ) {
      tags: nodes {
        name
        target {
          url: commitUrl
        }
      }
    }
  }
}
"""
 
 
class GitHubDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
    """Data update coordinator for the GitHub integration."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        client: GitHubAPI,
        repository: str,
    ) -> None:
        """Initialize GitHub data update coordinator base class."""
        self.repositoryrepository = repository
        self._client_client = client
        self._last_response_last_response: GitHubResponseModel[dict[str, Any]] | None = None
        self._subscription_id_subscription_id: str | None = None
        self.datadatadata = {}
 
        super().__init__(
            hass,
            LOGGER,
            name=repository,
            update_interval=FALLBACK_UPDATE_INTERVAL,
        )
 
    async def _async_update_data(self) -> GitHubResponseModel[dict[str, Any]]:
        """Update data."""
        owner, repository = self.repositoryrepository.split("/")
        try:
            response = await self._client_client.graphql(
                query=GRAPHQL_REPOSITORY_QUERY,
                variables={"owner": owner, "repository": repository},
            )
        except (GitHubConnectionException, GitHubRatelimitException) as exception:
            # These are expected and we dont log anything extra
            raise UpdateFailed(exception) from exception
        except GitHubException as exception:
            # These are unexpected and we log the trace to help with troubleshooting
            LOGGER.exception(exception)
            raise UpdateFailed(exception) from exception
 
        self._last_response_last_response = response
        return response.data["data"]["repository"]
 
    async def _handle_event(self, event: GitHubEventModel) -> None:
        """Handle an event."""
        if event.type in REFRESH_EVENT_TYPES:
            await self.async_request_refreshasync_request_refresh()
 
    @staticmethod
    async def _handle_error(error: GitHubException) -> None:
        """Handle an error."""
        LOGGER.error("An error occurred while processing new events - %s", error)
 
    async def subscribe(self) -> None:
        """Subscribe to repository events."""
        self._subscription_id_subscription_id = await self._client_client.repos.events.subscribe(
            self.repositoryrepository,
            event_callback=self._handle_event_handle_event,
            error_callback=self._handle_error_handle_error,
        )
        self.hasshass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self.unsubscribeunsubscribe)
 
    def unsubscribe(self, *args) -> None:
        """Unsubscribe to repository events."""
        self._client_client.repos.events.unsubscribe(subscription_id=self._subscription_id_subscription_id)