Move and rename remember_the_milk base entity to separate module (#126171)

Move remember_the_milk base entity to separate module
This commit is contained in:
epenet 2024-09-18 11:07:38 +02:00 committed by GitHub
parent df434fc5e2
commit dbb6eaa9eb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 149 additions and 137 deletions

View file

@ -4,17 +4,18 @@ import json
import logging
import os
from rtmapi import Rtm, RtmRequestFailedException
from rtmapi import Rtm
import voluptuous as vol
from homeassistant.components import configurator
from homeassistant.const import CONF_API_KEY, CONF_ID, CONF_NAME, CONF_TOKEN, STATE_OK
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.const import CONF_API_KEY, CONF_ID, CONF_NAME, CONF_TOKEN
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from .entity import RememberTheMilkEntity
# httplib2 is a transitive dependency from RtmAPI. If this dependency is not
# set explicitly, the library does not work.
_LOGGER = logging.getLogger(__name__)
@ -53,7 +54,7 @@ SERVICE_SCHEMA_COMPLETE_TASK = vol.Schema({vol.Required(CONF_ID): cv.string})
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Remember the milk component."""
component = EntityComponent[RememberTheMilk](_LOGGER, DOMAIN, hass)
component = EntityComponent[RememberTheMilkEntity](_LOGGER, DOMAIN, hass)
stored_rtm_config = RememberTheMilkConfiguration(hass)
for rtm_config in config[DOMAIN]:
@ -85,7 +86,7 @@ def setup(hass: HomeAssistant, config: ConfigType) -> bool:
def _create_instance(
hass, account_name, api_key, shared_secret, token, stored_rtm_config, component
):
entity = RememberTheMilk(
entity = RememberTheMilkEntity(
account_name, api_key, shared_secret, token, stored_rtm_config
)
component.add_entities([entity])
@ -237,134 +238,3 @@ class RememberTheMilkConfiguration:
if hass_id in self._config[profile_name][CONF_ID_MAP]:
del self._config[profile_name][CONF_ID_MAP][hass_id]
self.save_config()
class RememberTheMilk(Entity):
"""Representation of an interface to Remember The Milk."""
def __init__(self, name, api_key, shared_secret, token, rtm_config):
"""Create new instance of Remember The Milk component."""
self._name = name
self._api_key = api_key
self._shared_secret = shared_secret
self._token = token
self._rtm_config = rtm_config
self._rtm_api = Rtm(api_key, shared_secret, "delete", token)
self._token_valid = None
self._check_token()
_LOGGER.debug("Instance created for account %s", self._name)
def _check_token(self):
"""Check if the API token is still valid.
If it is not valid any more, delete it from the configuration. This
will trigger a new authentication process.
"""
valid = self._rtm_api.token_valid()
if not valid:
_LOGGER.error(
"Token for account %s is invalid. You need to register again!",
self.name,
)
self._rtm_config.delete_token(self._name)
self._token_valid = False
else:
self._token_valid = True
return self._token_valid
def create_task(self, call: ServiceCall) -> None:
"""Create a new task on Remember The Milk.
You can use the smart syntax to define the attributes of a new task,
e.g. "my task #some_tag ^today" will add tag "some_tag" and set the
due date to today.
"""
try:
task_name = call.data[CONF_NAME]
hass_id = call.data.get(CONF_ID)
rtm_id = None
if hass_id is not None:
rtm_id = self._rtm_config.get_rtm_id(self._name, hass_id)
result = self._rtm_api.rtm.timelines.create()
timeline = result.timeline.value
if hass_id is None or rtm_id is None:
result = self._rtm_api.rtm.tasks.add(
timeline=timeline, name=task_name, parse="1"
)
_LOGGER.debug(
"Created new task '%s' in account %s", task_name, self.name
)
self._rtm_config.set_rtm_id(
self._name,
hass_id,
result.list.id,
result.list.taskseries.id,
result.list.taskseries.task.id,
)
else:
self._rtm_api.rtm.tasks.setName(
name=task_name,
list_id=rtm_id[0],
taskseries_id=rtm_id[1],
task_id=rtm_id[2],
timeline=timeline,
)
_LOGGER.debug(
"Updated task with id '%s' in account %s to name %s",
hass_id,
self.name,
task_name,
)
except RtmRequestFailedException as rtm_exception:
_LOGGER.error(
"Error creating new Remember The Milk task for account %s: %s",
self._name,
rtm_exception,
)
def complete_task(self, call: ServiceCall) -> None:
"""Complete a task that was previously created by this component."""
hass_id = call.data[CONF_ID]
rtm_id = self._rtm_config.get_rtm_id(self._name, hass_id)
if rtm_id is None:
_LOGGER.error(
(
"Could not find task with ID %s in account %s. "
"So task could not be closed"
),
hass_id,
self._name,
)
return
try:
result = self._rtm_api.rtm.timelines.create()
timeline = result.timeline.value
self._rtm_api.rtm.tasks.complete(
list_id=rtm_id[0],
taskseries_id=rtm_id[1],
task_id=rtm_id[2],
timeline=timeline,
)
self._rtm_config.delete_rtm_id(self._name, hass_id)
_LOGGER.debug(
"Completed task with id %s in account %s", hass_id, self._name
)
except RtmRequestFailedException as rtm_exception:
_LOGGER.error(
"Error creating new Remember The Milk task for account %s: %s",
self._name,
rtm_exception,
)
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def state(self):
"""Return the state of the device."""
if not self._token_valid:
return "API token invalid"
return STATE_OK

View file

@ -0,0 +1,142 @@
"""Support to interact with Remember The Milk."""
import logging
from rtmapi import Rtm, RtmRequestFailedException
from homeassistant.const import CONF_ID, CONF_NAME, STATE_OK
from homeassistant.core import ServiceCall
from homeassistant.helpers.entity import Entity
_LOGGER = logging.getLogger(__name__)
class RememberTheMilkEntity(Entity):
"""Representation of an interface to Remember The Milk."""
def __init__(self, name, api_key, shared_secret, token, rtm_config):
"""Create new instance of Remember The Milk component."""
self._name = name
self._api_key = api_key
self._shared_secret = shared_secret
self._token = token
self._rtm_config = rtm_config
self._rtm_api = Rtm(api_key, shared_secret, "delete", token)
self._token_valid = None
self._check_token()
_LOGGER.debug("Instance created for account %s", self._name)
def _check_token(self):
"""Check if the API token is still valid.
If it is not valid any more, delete it from the configuration. This
will trigger a new authentication process.
"""
valid = self._rtm_api.token_valid()
if not valid:
_LOGGER.error(
"Token for account %s is invalid. You need to register again!",
self.name,
)
self._rtm_config.delete_token(self._name)
self._token_valid = False
else:
self._token_valid = True
return self._token_valid
def create_task(self, call: ServiceCall) -> None:
"""Create a new task on Remember The Milk.
You can use the smart syntax to define the attributes of a new task,
e.g. "my task #some_tag ^today" will add tag "some_tag" and set the
due date to today.
"""
try:
task_name = call.data[CONF_NAME]
hass_id = call.data.get(CONF_ID)
rtm_id = None
if hass_id is not None:
rtm_id = self._rtm_config.get_rtm_id(self._name, hass_id)
result = self._rtm_api.rtm.timelines.create()
timeline = result.timeline.value
if hass_id is None or rtm_id is None:
result = self._rtm_api.rtm.tasks.add(
timeline=timeline, name=task_name, parse="1"
)
_LOGGER.debug(
"Created new task '%s' in account %s", task_name, self.name
)
self._rtm_config.set_rtm_id(
self._name,
hass_id,
result.list.id,
result.list.taskseries.id,
result.list.taskseries.task.id,
)
else:
self._rtm_api.rtm.tasks.setName(
name=task_name,
list_id=rtm_id[0],
taskseries_id=rtm_id[1],
task_id=rtm_id[2],
timeline=timeline,
)
_LOGGER.debug(
"Updated task with id '%s' in account %s to name %s",
hass_id,
self.name,
task_name,
)
except RtmRequestFailedException as rtm_exception:
_LOGGER.error(
"Error creating new Remember The Milk task for account %s: %s",
self._name,
rtm_exception,
)
def complete_task(self, call: ServiceCall) -> None:
"""Complete a task that was previously created by this component."""
hass_id = call.data[CONF_ID]
rtm_id = self._rtm_config.get_rtm_id(self._name, hass_id)
if rtm_id is None:
_LOGGER.error(
(
"Could not find task with ID %s in account %s. "
"So task could not be closed"
),
hass_id,
self._name,
)
return
try:
result = self._rtm_api.rtm.timelines.create()
timeline = result.timeline.value
self._rtm_api.rtm.tasks.complete(
list_id=rtm_id[0],
taskseries_id=rtm_id[1],
task_id=rtm_id[2],
timeline=timeline,
)
self._rtm_config.delete_rtm_id(self._name, hass_id)
_LOGGER.debug(
"Completed task with id %s in account %s", hass_id, self._name
)
except RtmRequestFailedException as rtm_exception:
_LOGGER.error(
"Error creating new Remember The Milk task for account %s: %s",
self._name,
rtm_exception,
)
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def state(self):
"""Return the state of the device."""
if not self._token_valid:
return "API token invalid"
return STATE_OK