Add common module to ProxymoxVE integration (#141941)

add common module

homeassistant/components/proxmoxve/__init__.py

@@ -6,7 +6,6 @@ from datetime import timedelta
 from typing import Any
 
 from proxmoxer import AuthenticationError, ProxmoxAPI
-from proxmoxer.core import ResourceException
 import requests.exceptions
 from requests.exceptions import ConnectTimeout, SSLError
 import voluptuous as vol
@@ -25,6 +24,7 @@ from homeassistant.helpers.discovery import async_load_platform
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
+from .common import ProxmoxClient, call_api_container_vm, parse_api_container_vm
 from .const import (
     _LOGGER,
     CONF_CONTAINERS,
@@ -219,80 +219,3 @@ def create_coordinator_container_vm(
         update_method=async_update_data,
         update_interval=timedelta(seconds=UPDATE_INTERVAL),
     )
-
-
-def parse_api_container_vm(status: dict[str, Any]) -> dict[str, Any]:
-    """Get the container or vm api data and return it formatted in a dictionary.
-
-    It is implemented in this way to allow for more data to be added for sensors
-    in the future.
-    """
-
-    return {"status": status["status"], "name": status["name"]}
-
-
-def call_api_container_vm(
-    proxmox: ProxmoxAPI,
-    node_name: str,
-    vm_id: int,
-    machine_type: int,
-) -> dict[str, Any] | None:
-    """Make proper api calls."""
-    status = None
-
-    try:
-        if machine_type == TYPE_VM:
-            status = proxmox.nodes(node_name).qemu(vm_id).status.current.get()
-        elif machine_type == TYPE_CONTAINER:
-            status = proxmox.nodes(node_name).lxc(vm_id).status.current.get()
-    except (ResourceException, requests.exceptions.ConnectionError):
-        return None
-
-    return status
-
-
-class ProxmoxClient:
-    """A wrapper for the proxmoxer ProxmoxAPI client."""
-
-    _proxmox: ProxmoxAPI
-
-    def __init__(
-        self,
-        host: str,
-        port: int,
-        user: str,
-        realm: str,
-        password: str,
-        verify_ssl: bool,
-    ) -> None:
-        """Initialize the ProxmoxClient."""
-
-        self._host = host
-        self._port = port
-        self._user = user
-        self._realm = realm
-        self._password = password
-        self._verify_ssl = verify_ssl
-
-    def build_client(self) -> None:
-        """Construct the ProxmoxAPI client.
-
-        Allows inserting the realm within the `user` value.
-        """
-
-        if "@" in self._user:
-            user_id = self._user
-        else:
-            user_id = f"{self._user}@{self._realm}"
-
-        self._proxmox = ProxmoxAPI(
-            self._host,
-            port=self._port,
-            user=user_id,
-            password=self._password,
-            verify_ssl=self._verify_ssl,
-        )
-
-    def get_api_client(self) -> ProxmoxAPI:
-        """Return the ProxmoxAPI client."""
-        return self._proxmox

homeassistant/components/proxmoxve/common.py

@@ -0,0 +1,88 @@
+"""Commons for Proxmox VE integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from proxmoxer import ProxmoxAPI
+from proxmoxer.core import ResourceException
+import requests.exceptions
+
+from .const import TYPE_CONTAINER, TYPE_VM
+
+
+class ProxmoxClient:
+    """A wrapper for the proxmoxer ProxmoxAPI client."""
+
+    _proxmox: ProxmoxAPI
+
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        user: str,
+        realm: str,
+        password: str,
+        verify_ssl: bool,
+    ) -> None:
+        """Initialize the ProxmoxClient."""
+
+        self._host = host
+        self._port = port
+        self._user = user
+        self._realm = realm
+        self._password = password
+        self._verify_ssl = verify_ssl
+
+    def build_client(self) -> None:
+        """Construct the ProxmoxAPI client.
+
+        Allows inserting the realm within the `user` value.
+        """
+
+        if "@" in self._user:
+            user_id = self._user
+        else:
+            user_id = f"{self._user}@{self._realm}"
+
+        self._proxmox = ProxmoxAPI(
+            self._host,
+            port=self._port,
+            user=user_id,
+            password=self._password,
+            verify_ssl=self._verify_ssl,
+        )
+
+    def get_api_client(self) -> ProxmoxAPI:
+        """Return the ProxmoxAPI client."""
+        return self._proxmox
+
+
+def parse_api_container_vm(status: dict[str, Any]) -> dict[str, Any]:
+    """Get the container or vm api data and return it formatted in a dictionary.
+
+    It is implemented in this way to allow for more data to be added for sensors
+    in the future.
+    """
+
+    return {"status": status["status"], "name": status["name"]}
+
+
+def call_api_container_vm(
+    proxmox: ProxmoxAPI,
+    node_name: str,
+    vm_id: int,
+    machine_type: int,
+) -> dict[str, Any] | None:
+    """Make proper api calls."""
+    status = None
+
+    try:
+        if machine_type == TYPE_VM:
+            status = proxmox.nodes(node_name).qemu(vm_id).status.current.get()
+        elif machine_type == TYPE_CONTAINER:
+            status = proxmox.nodes(node_name).lxc(vm_id).status.current.get()
+    except (ResourceException, requests.exceptions.ConnectionError):
+        return None
+
+    return status