diff --git a/homeassistant/components/portainer/services.py b/homeassistant/components/portainer/services.py index ac878bad2669..70291632c2d9 100644 --- a/homeassistant/components/portainer/services.py +++ b/homeassistant/components/portainer/services.py @@ -1,6 +1,8 @@ """Services for the Portainer integration.""" +from collections.abc import Coroutine from datetime import timedelta +from typing import Any import probatio from pyportainer import ( @@ -117,18 +119,10 @@ def _async_get_container_and_endpoint_ids( ) -async def prune_images(call: ServiceCall) -> None: - """Prune unused images in Portainer, with more controls.""" - device, config_entry = _async_get_device_and_entry(call, call.data[ATTR_DEVICE_ID]) - coordinator = config_entry.runtime_data - endpoint_id = _async_get_endpoint_id(device, config_entry) - +async def _async_call_portainer(coroutine: Coroutine[Any, Any, Any]) -> None: + """Await a Portainer call, mapping library errors to HomeAssistantError.""" try: - await coordinator.portainer.images_prune( - endpoint_id=endpoint_id, - until=call.data.get(ATTR_DATE_UNTIL), - dangling=call.data.get(ATTR_DANGLING, False), - ) + await coroutine except PortainerAuthenticationError as err: raise HomeAssistantError( translation_domain=DOMAIN, @@ -146,33 +140,34 @@ async def prune_images(call: ServiceCall) -> None: ) from err +async def prune_images(call: ServiceCall) -> None: + """Prune unused images in Portainer, with more controls.""" + device, config_entry = _async_get_device_and_entry(call, call.data[ATTR_DEVICE_ID]) + coordinator = config_entry.runtime_data + endpoint_id = _async_get_endpoint_id(device, config_entry) + + await _async_call_portainer( + coordinator.portainer.images_prune( + endpoint_id=endpoint_id, + until=call.data.get(ATTR_DATE_UNTIL), + dangling=call.data.get(ATTR_DANGLING, False), + ) + ) + + async def prune_build_cache(call: ServiceCall) -> None: """Prune the build cache in Portainer, with more controls.""" device, config_entry = _async_get_device_and_entry(call, call.data[ATTR_DEVICE_ID]) coordinator = config_entry.runtime_data endpoint_id = _async_get_endpoint_id(device, config_entry) - try: - await coordinator.portainer.prune_build_cache( + await _async_call_portainer( + coordinator.portainer.prune_build_cache( endpoint_id, all_cache=call.data[ATTR_ALL], until=call.data.get(ATTR_DATE_UNTIL), ) - except PortainerAuthenticationError as err: - raise HomeAssistantError( - translation_domain=DOMAIN, - translation_key="invalid_auth", - ) from err - except PortainerConnectionError as err: - raise HomeAssistantError( - translation_domain=DOMAIN, - translation_key="cannot_connect", - ) from err - except PortainerTimeoutError as err: - raise HomeAssistantError( - translation_domain=DOMAIN, - translation_key="timeout_connect", - ) from err + ) async def recreate_container(call: ServiceCall) -> None: @@ -186,28 +181,14 @@ async def recreate_container(call: ServiceCall) -> None: ) timeout: timedelta | None = call.data.get(ATTR_TIMEOUT) - try: - await coordinator.portainer.container_recreate( + await _async_call_portainer( + coordinator.portainer.container_recreate( endpoint_id=endpoint_id, container_id=container_id, **({"timeout": timeout} if timeout is not None else {}), pull_image=call.data.get(ATTR_PULL_IMAGE, False), ) - except PortainerAuthenticationError as err: - raise HomeAssistantError( - translation_domain=DOMAIN, - translation_key="invalid_auth", - ) from err - except PortainerConnectionError as err: - raise HomeAssistantError( - translation_domain=DOMAIN, - translation_key="cannot_connect", - ) from err - except PortainerTimeoutError as err: - raise HomeAssistantError( - translation_domain=DOMAIN, - translation_key="timeout_connect", - ) from err + ) await coordinator.async_request_refresh()