Share the Portainer error helper with switches and buttons (#183996)

homeassistant/components/portainer/button.py

@@ -7,11 +7,6 @@ from datetime import timedelta
 from typing import Any, override
 
 from pyportainer import DockerContainerState, Portainer
-from pyportainer.exceptions import (
-    PortainerAuthenticationError,
-    PortainerConnectionError,
-    PortainerTimeoutError,
-)
 from pyportainer.models.docker import DockerContainer
 from pyportainer.models.stacks import Stack, StackType
 
@@ -22,11 +17,9 @@ from homeassistant.components.button import (
 )
 from homeassistant.const import EntityCategory
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
 from . import PortainerConfigEntry
-from .const import DOMAIN
 from .coordinator import (
     PortainerContainerData,
     PortainerCoordinator,
@@ -38,6 +31,7 @@ from .entity import (
     PortainerEndpointEntity,
     PortainerStackEntity,
 )
+from .util import async_call_portainer
 
 PARALLEL_UPDATES = 1
 
@@ -275,24 +269,7 @@ class PortainerBaseButton(ButtonEntity):
     @override
     async def async_press(self) -> None:
         """Trigger the Portainer button press service."""
-        try:
-            await self._async_press_call()
-        except PortainerConnectionError as err:
-            raise HomeAssistantError(
-                translation_domain=DOMAIN,
-                translation_key="cannot_connect",
-            ) from err
-        except PortainerAuthenticationError as err:
-            raise HomeAssistantError(
-                translation_domain=DOMAIN,
-                translation_key="invalid_auth",
-            ) from err
-        except PortainerTimeoutError as err:
-            raise HomeAssistantError(
-                translation_domain=DOMAIN,
-                translation_key="timeout_connect",
-            ) from err
-
+        await async_call_portainer(self._async_press_call())
         await self.coordinator.async_request_refresh()
 
 

homeassistant/components/portainer/services.py

@@ -1,20 +1,13 @@
 """Services for the Portainer integration."""
 
-from collections.abc import Coroutine
 from datetime import timedelta
 from enum import StrEnum
-from typing import Any
 
 import probatio
-from pyportainer import (
-    PortainerAuthenticationError,
-    PortainerConnectionError,
-    PortainerTimeoutError,
-)
 
 from homeassistant.const import ATTR_DEVICE_ID
 from homeassistant.core import HomeAssistant, ServiceCall, callback
-from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.exceptions import ServiceValidationError
 from homeassistant.helpers import (
     config_validation as cv,
     device_registry as dr,
@@ -23,6 +16,7 @@ from homeassistant.helpers import (
 
 from .const import DOMAIN
 from .coordinator import PortainerConfigEntry
+from .util import async_call_portainer
 
 
 class PortainerService(StrEnum):
@@ -130,34 +124,13 @@ def _async_get_container_and_endpoint_ids(
     )
 
 
-async def _async_call_portainer(coroutine: Coroutine[Any, Any, Any]) -> None:
-    """Await a Portainer call, mapping library errors to HomeAssistantError."""
-    try:
-        await coroutine
-    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 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(
+    await async_call_portainer(
         coordinator.portainer.images_prune(
             endpoint_id=endpoint_id,
             until=call.data.get(PortainerServiceArgument.UNTIL),
@@ -172,7 +145,7 @@ async def prune_build_cache(call: ServiceCall) -> None:
     coordinator = config_entry.runtime_data
     endpoint_id = _async_get_endpoint_id(device, config_entry)
 
-    await _async_call_portainer(
+    await async_call_portainer(
         coordinator.portainer.prune_build_cache(
             endpoint_id,
             all_cache=call.data[PortainerServiceArgument.ALL],
@@ -192,7 +165,7 @@ async def recreate_container(call: ServiceCall) -> None:
     )
     timeout: timedelta | None = call.data.get(PortainerServiceArgument.TIMEOUT)
 
-    await _async_call_portainer(
+    await async_call_portainer(
         coordinator.portainer.container_recreate(
             endpoint_id=endpoint_id,
             container_id=container_id,

homeassistant/components/portainer/switch.py

@@ -5,11 +5,6 @@ from dataclasses import dataclass
 from typing import Any, override
 
 from pyportainer import DockerContainerState, Portainer, StackStatus
-from pyportainer.exceptions import (
-    PortainerAuthenticationError,
-    PortainerConnectionError,
-    PortainerTimeoutError,
-)
 
 from homeassistant.components.switch import (
     SwitchDeviceClass,
@@ -17,11 +12,9 @@ from homeassistant.components.switch import (
     SwitchEntityDescription,
 )
 from homeassistant.core import HomeAssistant
-from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 
 from . import PortainerConfigEntry
-from .const import DOMAIN
 from .coordinator import (
     PortainerContainerData,
     PortainerCoordinator,
@@ -32,6 +25,7 @@ from .entity import (
     PortainerCoordinatorData,
     PortainerStackEntity,
 )
+from .util import async_call_portainer
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -60,25 +54,8 @@ async def _perform_action(
     coroutine: Coroutine[Any, Any, Any],
 ) -> None:
     """Perform a Portainer action with error handling and coordinator refresh."""
-    try:
-        await coroutine
-    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
-    else:
-        await coordinator.async_request_refresh()
+    await async_call_portainer(coroutine)
+    await coordinator.async_request_refresh()
 
 
 CONTAINER_SWITCHES: tuple[PortainerSwitchEntityDescription, ...] = (

homeassistant/components/portainer/util.py

@@ -1,6 +1,40 @@
 """Utility functions for the Portainer integration."""
 
+from collections.abc import Coroutine
+from typing import Any
+
+from pyportainer import (
+    PortainerAuthenticationError,
+    PortainerConnectionError,
+    PortainerTimeoutError,
+)
+
+from homeassistant.exceptions import HomeAssistantError
+
+from .const import DOMAIN
+
 
 def sanitize_container_name(container_name: str) -> str:
     """Sanitize to get a proper container name."""
     return container_name.replace("/", " ").strip()
+
+
+async def async_call_portainer(coroutine: Coroutine[Any, Any, Any]) -> None:
+    """Await a Portainer call, mapping library errors to HomeAssistantError."""
+    try:
+        await coroutine
+    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