refactor: modernize floating_ip and primary_ip modules (#771)

##### SUMMARY

Use the latest best practices for modules, and prepare for the upcoming
changes.

plugins/module_utils/floating_ip.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from ..module_utils.vendor.hcloud.floating_ips import (
+    BoundFloatingIP,
+)
+
+
+def prepare_result(o: BoundFloatingIP):
+    return {
+        "id": o.id,
+        "name": o.name,
+        "description": o.description,
+        "ip": o.ip,
+        "type": o.type,
+        "home_location": o.home_location.name,
+        "labels": o.labels,
+        "server": o.server.name if o.server is not None else None,
+        "delete_protection": o.protection["delete"],
+    }

plugins/module_utils/primary_ip.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from ..module_utils.vendor.hcloud.primary_ips import (
+    BoundPrimaryIP,
+)
+
+
+def prepare_result(o: BoundPrimaryIP):
+    return {
+        "id": o.id,
+        "name": o.name,
+        "ip": o.ip,
+        "type": o.type,
+        "datacenter": o.datacenter.name,
+        "labels": o.labels,
+        "delete_protection": o.protection["delete"],
+        "assignee_id": o.assignee_id,
+        "assignee_type": o.assignee_type,
+        "auto_delete": o.auto_delete,
+    }

plugins/modules/floating_ip.py

@@ -160,142 +160,155 @@ hcloud_floating_ip:
                 mylabel: 123
 """
 
-from ansible.module_utils.basic import AnsibleModule
+from ..module_utils import floating_ip
-
+from ..module_utils.hcloud import AnsibleHCloud, AnsibleModule
-from ..module_utils.hcloud import AnsibleHCloud
 from ..module_utils.vendor.hcloud import HCloudException
 from ..module_utils.vendor.hcloud.floating_ips import BoundFloatingIP
 
 
-class AnsibleHCloudFloatingIP(AnsibleHCloud):
+class AnsibleFloatingIP(AnsibleHCloud):
-    represent = "hcloud_floating_ip"
+    represent = "floating_ip"
 
-    hcloud_floating_ip: BoundFloatingIP | None = None
+    floating_ip: BoundFloatingIP | None = None
 
     def _prepare_result(self):
-        return {
+        if self.floating_ip is None:
-            "id": self.hcloud_floating_ip.id,
+            return {}
-            "name": self.hcloud_floating_ip.name,
+        return floating_ip.prepare_result(self.floating_ip)
-            "description": self.hcloud_floating_ip.description,
+
-            "ip": self.hcloud_floating_ip.ip,
+    def _get(self):
-            "type": self.hcloud_floating_ip.type,
+        if (value := self.module.params.get("id")) is not None:
-            "home_location": self.hcloud_floating_ip.home_location.name,
+            self.floating_ip = self.client.floating_ips.get_by_id(value)
-            "labels": self.hcloud_floating_ip.labels,
+        elif (value := self.module.params.get("name")) is not None:
-            "server": self.hcloud_floating_ip.server.name if self.hcloud_floating_ip.server is not None else None,
+            self.floating_ip = self.client.floating_ips.get_by_name(value)
-            "delete_protection": self.hcloud_floating_ip.protection["delete"],
+
+    def _create(self):
+        self.fail_on_invalid_params(
+            required=["name", "type"],
+            required_one_of=[["home_location", "server"]],
+        )
+
+        params = {
+            "name": self.module.params.get("name"),
+            "type": self.module.params.get("type"),
         }
 
-    def _get_floating_ip(self):
+        if (value := self.module.params.get("home_location")) is not None:
-        try:
+            params["home_location"] = self.client.locations.get_by_name(value)
-            if self.module.params.get("id") is not None:
+        elif (value := self.module.params.get("server")) is not None:
-                self.hcloud_floating_ip = self.client.floating_ips.get_by_id(self.module.params.get("id"))
+            params["server"] = self.client.servers.get_by_name(value)
-            else:
-                self.hcloud_floating_ip = self.client.floating_ips.get_by_name(self.module.params.get("name"))
-        except HCloudException as exception:
-            self.fail_json_hcloud(exception)
-
-    def _create_floating_ip(self):
-        self.module.fail_on_missing_params(required_params=["type"])
-        try:
-            params = {
-                "description": self.module.params.get("description"),
-                "type": self.module.params.get("type"),
-                "name": self.module.params.get("name"),
-            }
-            if self.module.params.get("home_location") is not None:
-                params["home_location"] = self.client.locations.get_by_name(self.module.params.get("home_location"))
-            elif self.module.params.get("server") is not None:
-                params["server"] = self.client.servers.get_by_name(self.module.params.get("server"))
-            else:
-                self.module.fail_json(msg="one of the following is required: home_location, server")
-
-            if self.module.params.get("labels") is not None:
-                params["labels"] = self.module.params.get("labels")
-            if not self.module.check_mode:
-                resp = self.client.floating_ips.create(**params)
-                self.hcloud_floating_ip = resp.floating_ip
-
-                delete_protection = self.module.params.get("delete_protection")
-                if delete_protection is not None:
-                    action = self.hcloud_floating_ip.change_protection(delete=delete_protection)
-                    action.wait_until_finished()
-        except HCloudException as exception:
-            self.fail_json_hcloud(exception)
-        self._mark_as_changed()
-        self._get_floating_ip()
-
-    def _update_floating_ip(self):
-        try:
-            labels = self.module.params.get("labels")
-            if labels is not None and labels != self.hcloud_floating_ip.labels:
-                if not self.module.check_mode:
-                    self.hcloud_floating_ip.update(labels=labels)
-                self._mark_as_changed()
-
-            description = self.module.params.get("description")
-            if description is not None and description != self.hcloud_floating_ip.description:
-                if not self.module.check_mode:
-                    self.hcloud_floating_ip.update(description=description)
-                self._mark_as_changed()
-
-            server = self.module.params.get("server")
-            if server is not None and self.hcloud_floating_ip.server is not None:
-                if self.module.params.get("force") and server != self.hcloud_floating_ip.server.name:
-                    if not self.module.check_mode:
-                        action = self.hcloud_floating_ip.assign(self.client.servers.get_by_name(server))
-                        action.wait_until_finished()
-                        self._mark_as_changed()
-                elif server != self.hcloud_floating_ip.server.name:
-                    self.module.warn(
-                        "Floating IP is already assigned to another server "
-                        f"{self.hcloud_floating_ip.server.name}. You need to "
-                        "unassign the Floating IP or use force=true."
-                    )
-                    self._mark_as_changed()
-            elif server is not None and self.hcloud_floating_ip.server is None:
-                if not self.module.check_mode:
-                    action = self.hcloud_floating_ip.assign(self.client.servers.get_by_name(server))
-                    action.wait_until_finished()
-                self._mark_as_changed()
-            elif server is None and self.hcloud_floating_ip.server is not None:
-                if not self.module.check_mode:
-                    action = self.hcloud_floating_ip.unassign()
-                    action.wait_until_finished()
-                self._mark_as_changed()
-
-            delete_protection = self.module.params.get("delete_protection")
-            if delete_protection is not None and delete_protection != self.hcloud_floating_ip.protection["delete"]:
-                if not self.module.check_mode:
-                    action = self.hcloud_floating_ip.change_protection(delete=delete_protection)
-                    action.wait_until_finished()
-                self._mark_as_changed()
-
-            self._get_floating_ip()
-        except HCloudException as exception:
-            self.fail_json_hcloud(exception)
-
-    def present_floating_ip(self):
-        self._get_floating_ip()
-        if self.hcloud_floating_ip is None:
-            self._create_floating_ip()
         else:
-            self._update_floating_ip()
+            self.module.fail_json(msg="one of the following is required: home_location, server")
 
-    def delete_floating_ip(self):
+        if (value := self.module.params.get("description")) is not None:
-        try:
+            params["description"] = value
-            self._get_floating_ip()
+
-            if self.hcloud_floating_ip is not None:
+        if (value := self.module.params.get("labels")) is not None:
-                if self.module.params.get("force") or self.hcloud_floating_ip.server is None:
+            params["labels"] = value
-                    if not self.module.check_mode:
+
-                        self.client.floating_ips.delete(self.hcloud_floating_ip)
+        if not self.module.check_mode:
-                else:
+            resp = self.client.floating_ips.create(**params)
-                    self.module.warn(
+            self.floating_ip = resp.floating_ip
-                        "Floating IP is currently assigned to server "
+            if resp.action is not None:
-                        f"{self.hcloud_floating_ip.server.name}. You need to "
+                resp.action.wait_until_finished()
-                        "unassign the Floating IP or use force=true."
+        self._mark_as_changed()
-                    )
+
+        if (value := self.module.params.get("delete_protection")) is not None:
+            if not self.module.check_mode:
+                action = self.floating_ip.change_protection(delete=value)
+                action.wait_until_finished()
+            self._mark_as_changed()
+
+        if not self.module.check_mode:
+            self.floating_ip.reload()
+
+    def _update(self):
+        need_reload = False
+
+        if (value := self.module.params.get("delete_protection")) is not None:
+            if value != self.floating_ip.protection["delete"]:
+                if not self.module.check_mode:
+                    action = self.floating_ip.change_protection(delete=value)
+                    action.wait_until_finished()
+                    need_reload = True
                 self._mark_as_changed()
-            self.hcloud_floating_ip = None
+
+        if (value := self.module.params.get("server")) is not None:
+            if self.floating_ip.server is not None:
+                if value != self.floating_ip.server.name:
+                    if self.module.params.get("force"):
+                        if not self.module.check_mode:
+                            action = self.floating_ip.assign(self.client.servers.get_by_name(value))
+                            action.wait_until_finished()
+                            need_reload = True
+                        self._mark_as_changed()
+                    else:
+                        self.module.warn(
+                            "Floating IP is already assigned to another server "
+                            f"{self.floating_ip.server.name}. You need to "
+                            "unassign the Floating IP or use force=true."
+                        )
+
+            else:  # self.floating_ip.server is None
+                if not self.module.check_mode:
+                    action = self.floating_ip.assign(self.client.servers.get_by_name(value))
+                    action.wait_until_finished()
+                    need_reload = True
+                self._mark_as_changed()
+
+        else:  # value is None
+            if self.floating_ip.server is not None:
+                if not self.module.check_mode:
+                    action = self.floating_ip.unassign()
+                    action.wait_until_finished()
+                    need_reload = True
+                self._mark_as_changed()
+
+        params = {}
+
+        if (value := self.module.params.get("labels")) is not None:
+            if value != self.floating_ip.labels:
+                params["labels"] = value
+                self._mark_as_changed()
+
+        if (value := self.module.params.get("description")) is not None:
+            if value != self.floating_ip.description:
+                params["description"] = value
+                self._mark_as_changed()
+
+        if params or need_reload:
+            if not self.module.check_mode:
+                self.floating_ip = self.floating_ip.update(**params)
+
+    def _delete(self):
+        if self.module.params.get("force") or self.floating_ip.server is None:
+            if not self.module.check_mode:
+                self.floating_ip.delete()
+            self._mark_as_changed()
+        else:
+            self.module.warn(
+                "Floating IP is currently assigned to server "
+                f"{self.floating_ip.server.name}. You need to "
+                "unassign the Floating IP or use force=true."
+            )
+
+        self.floating_ip = None
+
+    def present(self):
+        try:
+            self._get()
+            if self.floating_ip is None:
+                self._create()
+            else:
+                self._update()
+        except HCloudException as exception:
+            self.fail_json_hcloud(exception)
+
+    def delete(self):
+        try:
+            self._get()
+            if self.floating_ip is not None:
+                self._delete()
         except HCloudException as exception:
             self.fail_json_hcloud(exception)
 
@@ -325,16 +338,18 @@ class AnsibleHCloudFloatingIP(AnsibleHCloud):
 
 
 def main():
-    module = AnsibleHCloudFloatingIP.define_module()
+    o = AnsibleFloatingIP(AnsibleFloatingIP.define_module())
 
-    hcloud = AnsibleHCloudFloatingIP(module)
+    match o.module.params["state"]:
-    state = module.params["state"]
+        case "absent":
-    if state == "absent":
+            o.delete()
-        hcloud.delete_floating_ip()
+        case "present":
-    elif state == "present":
+            o.present()
-        hcloud.present_floating_ip()
 
-    module.exit_json(**hcloud.get_result())
+    result = o.get_result()
+    result["hcloud_floating_ip"] = result.pop(o.represent)
+
+    o.module.exit_json(**result)
 
 
 if __name__ == "__main__":

plugins/modules/primary_ip.py

@@ -168,120 +168,116 @@ hcloud_primary_ip:
             sample: false
 """
 
-from ansible.module_utils.basic import AnsibleModule
+from typing import TYPE_CHECKING
 
-from ..module_utils.hcloud import AnsibleHCloud
+from ..module_utils import primary_ip
+from ..module_utils.hcloud import AnsibleHCloud, AnsibleModule
 from ..module_utils.vendor.hcloud import HCloudException
 from ..module_utils.vendor.hcloud.primary_ips import BoundPrimaryIP
 
+if TYPE_CHECKING:
+    from ..module_utils.vendor.hcloud.servers import BoundServer
 
-class AnsibleHCloudPrimaryIP(AnsibleHCloud):
-    represent = "hcloud_primary_ip"
 
-    hcloud_primary_ip: BoundPrimaryIP | None = None
+class AnsiblePrimaryIP(AnsibleHCloud):
+    represent = "primary_ip"
+
+    primary_ip: BoundPrimaryIP | None = None
 
     def _prepare_result(self):
-        return {
+        if self.primary_ip is None:
-            "id": self.hcloud_primary_ip.id,
+            return {}
-            "name": self.hcloud_primary_ip.name,
+        return primary_ip.prepare_result(self.primary_ip)
-            "ip": self.hcloud_primary_ip.ip,
-            "type": self.hcloud_primary_ip.type,
-            "datacenter": self.hcloud_primary_ip.datacenter.name,
-            "labels": self.hcloud_primary_ip.labels,
-            "delete_protection": self.hcloud_primary_ip.protection["delete"],
-            "assignee_id": (
-                self.hcloud_primary_ip.assignee_id if self.hcloud_primary_ip.assignee_id is not None else None
-            ),
-            "assignee_type": self.hcloud_primary_ip.assignee_type,
-            "auto_delete": self.hcloud_primary_ip.auto_delete,
-        }
 
-    def _get_primary_ip(self):
+    def _get(self):
-        try:
+        if (value := self.module.params.get("id")) is not None:
-            if self.module.params.get("id") is not None:
+            self.primary_ip = self.client.primary_ips.get_by_id(value)
-                self.hcloud_primary_ip = self.client.primary_ips.get_by_id(self.module.params.get("id"))
+        elif (value := self.module.params.get("name")) is not None:
-            else:
+            self.primary_ip = self.client.primary_ips.get_by_name(value)
-                self.hcloud_primary_ip = self.client.primary_ips.get_by_name(self.module.params.get("name"))
-        except HCloudException as exception:
-            self.fail_json_hcloud(exception)
 
-    def _create_primary_ip(self):
+    def _create(self):
         self.fail_on_invalid_params(
-            required=["type", "name"],
+            required=["name", "type"],
             required_one_of=[["server", "datacenter"]],
         )
-        try:
+        params = {
-            params = {
+            "name": self.module.params.get("name"),
-                "type": self.module.params.get("type"),
+            "type": self.module.params.get("type"),
-                "name": self.module.params.get("name"),
+        }
-                "auto_delete": self.module.params.get("auto_delete"),
-            }
 
-            if self.module.params.get("datacenter") is not None:
+        if (value := self.module.params.get("datacenter")) is not None:
-                params["datacenter"] = self.client.datacenters.get_by_name(self.module.params.get("datacenter"))
+            params["datacenter"] = self.client.datacenters.get_by_name(value)
-            elif self.module.params.get("server") is not None:
+        elif (value := self.module.params.get("server")) is not None:
-                params["assignee_id"] = self._client_get_by_name_or_id("servers", self.module.params.get("server")).id
+            server: BoundServer = self._client_get_by_name_or_id("servers", value)
+            params["assignee_id"] = server.id
 
-            if self.module.params.get("labels") is not None:
+        if (value := self.module.params.get("auto_delete")) is not None:
-                params["labels"] = self.module.params.get("labels")
+            params["auto_delete"] = value
-            if not self.module.check_mode:
-                resp = self.client.primary_ips.create(**params)
-                if resp.action is not None:
-                    resp.action.wait_until_finished()
-                self.hcloud_primary_ip = resp.primary_ip
 
-                delete_protection = self.module.params.get("delete_protection")
+        if (value := self.module.params.get("labels")) is not None:
-                if delete_protection is not None:
+            params["labels"] = value
-                    action = self.hcloud_primary_ip.change_protection(delete=delete_protection)
+
-                    action.wait_until_finished()
+        if not self.module.check_mode:
-        except HCloudException as exception:
+            resp = self.client.primary_ips.create(**params)
-            self.fail_json_hcloud(exception)
+            if resp.action is not None:
+                resp.action.wait_until_finished()
+            self.primary_ip = resp.primary_ip
         self._mark_as_changed()
-        self._get_primary_ip()
 
-    def _update_primary_ip(self):
+        if (value := self.module.params.get("delete_protection")) is not None:
-        try:
+            if not self.module.check_mode:
-            changes = {}
+                action = self.primary_ip.change_protection(delete=value)
+                action.wait_until_finished()
+            self._mark_as_changed()
 
-            auto_delete = self.module.params.get("auto_delete")
+        if not self.module.check_mode:
-            if auto_delete is not None and auto_delete != self.hcloud_primary_ip.auto_delete:
+            self.primary_ip.reload()
-                changes["auto_delete"] = auto_delete
 
-            labels = self.module.params.get("labels")
+    def _update(self):
-            if labels is not None and labels != self.hcloud_primary_ip.labels:
+        need_reload = True
-                changes["labels"] = labels
 
-            if changes:
+        if (value := self.module.params.get("delete_protection")) is not None:
+            if value != self.primary_ip.protection["delete"]:
                 if not self.module.check_mode:
-                    self.hcloud_primary_ip.update(**changes)
+                    action = self.primary_ip.change_protection(delete=value)
-                self._mark_as_changed()
-
-            delete_protection = self.module.params.get("delete_protection")
-            if delete_protection is not None and delete_protection != self.hcloud_primary_ip.protection["delete"]:
-                if not self.module.check_mode:
-                    action = self.hcloud_primary_ip.change_protection(delete=delete_protection)
                     action.wait_until_finished()
+                    need_reload = True
                 self._mark_as_changed()
 
-            self._get_primary_ip()
+        params = {}
+
+        if (value := self.module.params.get("auto_delete")) is not None:
+            if value != self.primary_ip.auto_delete:
+                params["auto_delete"] = value
+                self._mark_as_changed()
+
+        if (value := self.module.params.get("labels")) is not None:
+            if value != self.primary_ip.labels:
+                params["labels"] = value
+                self._mark_as_changed()
+
+        if params or need_reload:
+            if not self.module.check_mode:
+                self.primary_ip = self.primary_ip.update(**params)
+
+    def present(self):
+        try:
+            self._get()
+            if self.primary_ip is None:
+                self._create()
+            else:
+                self._update()
         except HCloudException as exception:
             self.fail_json_hcloud(exception)
 
-    def present_primary_ip(self):
+    def delete(self):
-        self._get_primary_ip()
-        if self.hcloud_primary_ip is None:
-            self._create_primary_ip()
-        else:
-            self._update_primary_ip()
-
-    def delete_primary_ip(self):
         try:
-            self._get_primary_ip()
+            self._get()
-            if self.hcloud_primary_ip is not None:
+            if self.primary_ip is not None:
                 if not self.module.check_mode:
-                    self.client.primary_ips.delete(self.hcloud_primary_ip)
+                    self.primary_ip.delete()
                 self._mark_as_changed()
-            self.hcloud_primary_ip = None
+            self.primary_ip = None
         except HCloudException as exception:
             self.fail_json_hcloud(exception)
 
@@ -309,16 +305,18 @@ class AnsibleHCloudPrimaryIP(AnsibleHCloud):
 
 
 def main():
-    module = AnsibleHCloudPrimaryIP.define_module()
+    o = AnsiblePrimaryIP(AnsiblePrimaryIP.define_module())
 
-    hcloud = AnsibleHCloudPrimaryIP(module)
+    match o.module.params["state"]:
-    state = module.params["state"]
+        case "absent":
-    if state == "absent":
+            o.delete()
-        hcloud.delete_primary_ip()
+        case "present":
-    elif state == "present":
+            o.present()
-        hcloud.present_primary_ip()
 
-    module.exit_json(**hcloud.get_result())
+    result = o.get_result()
+    result["hcloud_primary_ip"] = result.pop(o.represent)
+
+    o.module.exit_json(**result)
 
 
 if __name__ == "__main__":

tests/integration/targets/floating_ip/tasks/test.yml

@@ -312,7 +312,7 @@
 - name: verify test already assigned Floating IP assign without force
   assert:
     that:
-      - floatingIP is changed
+      - floatingIP is not changed
       - floatingIP.hcloud_floating_ip.server == main_server.hcloud_server.name
 
 - name: test already assigned Floating IP assign with force