coordinator.py

Go to the documentation of this file.

"""DataUpdateCoordinator for the Todoist component."""
 
from datetime import timedelta
import logging
 
from todoist_api_python.api_async import TodoistAPIAsync
from todoist_api_python.models import Label, Project, Section, Task
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 
class TodoistCoordinator(DataUpdateCoordinator[list[Task]]):
    """Coordinator for updating task data from Todoist."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        logger: logging.Logger,
        entry: ConfigEntry | None,
        update_interval: timedelta,
        api: TodoistAPIAsync,
        token: str,
    ) -> None:
        """Initialize the Todoist coordinator."""
        super().__init__(
            hass,
            logger,
            config_entry=entry,
            name="Todoist",
            update_interval=update_interval,
        )
        self.apiapi = api
        self._projects_projects: list[Project] | None = None
        self._labels_labels: list[Label] | None = None
        self.tokentoken = token
 
    async def _async_update_data(self) -> list[Task]:
        """Fetch tasks from the Todoist API."""
        try:
            return await self.apiapi.get_tasks()
        except Exception as err:
            raise UpdateFailed(f"Error communicating with API: {err}") from err
 
    async def async_get_projects(self) -> list[Project]:
        """Return todoist projects fetched at most once."""
        if self._projects_projects is None:
            self._projects_projects = await self.apiapi.get_projects()
        return self._projects_projects
 
    async def async_get_sections(self, project_id: str) -> list[Section]:
        """Return todoist sections for a given project ID."""
        return await self.apiapi.get_sections(project_id=project_id)
 
    async def async_get_labels(self) -> list[Label]:
        """Return todoist labels fetched at most once."""
        if self._labels_labels is None:
            self._labels_labels = await self.apiapi.get_labels()
        return self._labels_labels