Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def remove_apps(token: str):
"""Remove Home Assistant apps and installed apps."""
async with aiohttp.ClientSession() as session:
api = SmartThings(session, token)
apps = await api.apps()
installed_apps = await api.installed_apps()
for app in apps:
if not app.app_name.startswith('homeassistant.'):
continue
# Remove installed apps first
for installed_app in installed_apps:
if installed_app.app_id == app.app_id:
await api.delete_installed_app(
installed_app.installed_app_id)
print("Removed installed app '{}' ({})".format(
installed_app.display_name,
installed_app.installed_app_id))
# Remove the app itself
async def async_remove_entry(hass: HomeAssistantType, entry: ConfigEntry) -> None:
"""Perform clean-up when entry is being removed."""
api = SmartThings(async_get_clientsession(hass), entry.data[CONF_ACCESS_TOKEN])
# Remove the installed_app, which if already removed raises a 403 error.
installed_app_id = entry.data[CONF_INSTALLED_APP_ID]
try:
await api.delete_installed_app(installed_app_id)
except ClientResponseError as ex:
if ex.status == 403:
_LOGGER.debug(
"Installed app %s has already been removed",
installed_app_id,
exc_info=True,
)
else:
raise
_LOGGER.debug("Removed installed app %s", installed_app_id)
async def async_step_install(self, data=None):
"""
Create a config entry at completion of a flow.
Launched when the user completes the flow or when the SmartApp
is installed into an additional location.
"""
if not self.api:
# Launched from the SmartApp install event handler
self.api = SmartThings(
async_get_clientsession(self.hass), data[CONF_ACCESS_TOKEN]
)
location = await self.api.location(data[CONF_LOCATION_ID])
return self.async_create_entry(title=location.name, data=data)
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
"""Initialize config entry which represents an installed SmartApp."""
if not validate_webhook_requirements(hass):
_LOGGER.warning(
"The 'base_url' of the 'http' integration must be configured and start with 'https://'"
)
return False
api = SmartThings(async_get_clientsession(hass), entry.data[CONF_ACCESS_TOKEN])
remove_entry = False
try:
# See if the app is already setup. This occurs when there are
# installs in multiple SmartThings locations (valid use-case)
manager = hass.data[DOMAIN][DATA_MANAGER]
smart_app = manager.smartapps.get(entry.data[CONF_APP_ID])
if not smart_app:
# Validate and setup the app.
app = await api.app(entry.data[CONF_APP_ID])
smart_app = setup_smartapp(hass, app)
# Validate and retrieve the installed app.
installed_app = await validate_installed_app(
api, entry.data[CONF_INSTALLED_APP_ID]
)
async def async_step_user(self, user_input=None):
"""Get access token and validate it."""
errors = {}
if user_input is None or CONF_ACCESS_TOKEN not in user_input:
return self._show_step_user(errors)
self.access_token = user_input.get(CONF_ACCESS_TOKEN, "")
self.api = SmartThings(async_get_clientsession(self.hass), self.access_token)
# Ensure token is a UUID
if not VAL_UID_MATCHER.match(self.access_token):
errors[CONF_ACCESS_TOKEN] = "token_invalid_format"
return self._show_step_user(errors)
# Check not already setup in another entry
if any(
entry.data.get(CONF_ACCESS_TOKEN) == self.access_token
for entry in self.hass.config_entries.async_entries(DOMAIN)
):
errors[CONF_ACCESS_TOKEN] = "token_already_setup"
return self._show_step_user(errors)
# Setup end-point
await setup_smartapp_endpoint(self.hass)
async def smartapp_sync_subscriptions(
hass: HomeAssistantType,
auth_token: str,
location_id: str,
installed_app_id: str,
devices,
):
"""Synchronize subscriptions of an installed up."""
api = SmartThings(async_get_clientsession(hass), auth_token)
tasks = []
async def create_subscription(target: str):
sub = Subscription()
sub.installed_app_id = installed_app_id
sub.location_id = location_id
sub.source_type = SourceType.CAPABILITY
sub.capability = target
try:
await api.create_subscription(sub)
_LOGGER.debug(
"Created subscription for '%s' under app '%s'", target, installed_app_id
)
except Exception as error: # pylint:disable=broad-except
_LOGGER.error(
"Failed to create subscription for '%s' under app '%s': %s",