Refactor(eos_designs): Optimize eos_designs_structured_config file wr…

…ite (#4283)

ansible_collections/arista/avd/plugins/action/eos_designs_structured_config.py

@@ -2,10 +2,11 @@
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the LICENSE file.
 
-
 import cProfile
+import json
 import pstats
 from collections import ChainMap
+from pathlib import Path
 from typing import Any
 
 import yaml
@@ -45,7 +46,8 @@ def run(self, tmp: Any = None, task_vars: dict | None = None) -> dict:
             profiler.enable()
 
         eos_designs_custom_templates = self._task.args.get("eos_designs_custom_templates", [])
-        self.dest = self._task.args.get("dest", False)
+        filename = str(self._task.args.get("dest", ""))
+        file_mode = str(self._task.args.get("mode", "0o664"))
         template_output = self._task.args.get("template_output", False)
         validation_mode = self._task.args.get("validation_mode")
 
@@ -142,19 +144,23 @@ def run(self, tmp: Any = None, task_vars: dict | None = None) -> dict:
             with self._templar.set_temporary_context(available_variables=template_vars):
                 output = self._templar.template(output, fail_on_undefined=False)
 
-        # If the argument 'dest' is set, write the output data to a file.
-        if self.dest:
-            # Depending on the file suffix of 'dest' (default: 'json') we will format the data to yaml or just write the output data directly.
-            # The Copy module used in 'write_file' will convert the output data to json automatically.
-            if self.dest.split(".")[-1] in ["yml", "yaml"]:
-                write_file_result = self.write_file(yaml.dump(output, Dumper=AnsibleDumper, indent=2, sort_keys=False, width=130), task_vars)
+        # If the argument 'dest' (filename) is set, write the output data to a file.
+        if filename:
+            # Depending on the file suffix of 'filename' (default: 'json') we will format the data to yaml or just write the output data directly.
+            if filename.endswith((".yml", ".yaml")):
+                result["changed"] = self.write_file(
+                    content=yaml.dump(output, Dumper=AnsibleDumper, indent=2, sort_keys=False, width=130),
+                    filename=filename,
+                    file_mode=file_mode,
+                )
             else:
-                write_file_result = self.write_file(output, task_vars)
-
-            # Overwrite result with the result from the copy operation (setting 'changed' flag accordingly)
-            result.update(write_file_result)
+                result["changed"] = self.write_file(
+                    content=json.dumps(output),
+                    filename=filename,
+                    file_mode=file_mode,
+                )
 
-        # If 'dest' is not set, hardcode 'changed' to true, since we don't know if something changed and later tasks may depend on this.
+        # If 'dest' (filename) is not set, hardcode 'changed' to true, since we don't know if something changed and later tasks may depend on this.
         else:
             result["changed"] = True
 
@@ -168,27 +174,27 @@ def run(self, tmp: Any = None, task_vars: dict | None = None) -> dict:
 
         return result
 
-    def write_file(self, content: str, task_vars: dict) -> dict:
+    def write_file(self, content: str, filename: str, file_mode: str = "0o664", dir_mode: str = "0o775") -> bool:
         """
-        This function implements the Ansible 'copy' action_module, to benefit from Ansible builtin functionality like 'changed'.
+        This function writes the file only if the content has changed.
 
-        Reuse task data.
-        """
-        new_task = self._task.copy()
-        new_task.args = {
-            "dest": self.dest,
-            "mode": self._task.args.get("mode"),
-            "content": content,
-        }
-
-        copy_action = self._shared_loader_obj.action_loader.get(
-            "ansible.legacy.copy",
-            task=new_task,
-            connection=self._connection,
-            play_context=self._play_context,
-            loader=self._loader,
-            templar=self._templar,
-            shared_loader_obj=self._shared_loader_obj,
-        )
+        Parameters
+        ----------
+            content: The content to write
+            filename: Target filename
 
-        return copy_action.run(task_vars=task_vars)
+        Returns:
+        -------
+            bool: Indicate if the content of filename has changed.
+        """
+        path = Path(filename)
+        if not path.exists():
+            # Create parent dirs automatically.
+            path.parent.mkdir(mode=int(dir_mode, 8), parents=True, exist_ok=True)
+            # Touch file
+            path.touch(mode=int(file_mode, 8))
+        elif path.read_text(encoding="UTF-8") == content:
+            return False
+
+        path.write_text(content, encoding="UTF-8")
+        return True