Add an option to update resources which were generated by class_generator

.flake8

@@ -52,6 +52,7 @@ fcn_exclude_functions =
     meta,
     tmp_path_factory,
     tmpdir_factory,
+    Path,
 
 enable-extensions =
     FCN,

class_generator/README.md

@@ -48,6 +48,14 @@ class-generator --interactive
 class-generator --kind Pod --add-tests
 ```
 
+### Updating existing resources
+
+To update existing resources which were generated with the script, pass `--update-resources`
+
+```bash
+class-generator --update-resources
+```
+
 ## Reporting an issue
 
 - Running with debug mode and `--debug` flag:

class_generator/class_generator.py

@@ -1,16 +1,26 @@
 from __future__ import annotations
+
+import filecmp
 import json
 import shlex
 import os
 import sys
+from concurrent.futures import Future, as_completed
+from concurrent.futures.thread import ThreadPoolExecutor
 from pathlib import Path
 
 from typing import Any, Dict, List, Optional
 import click
 import re
 
 import cloup
-from cloup.constraints import If, accept_none, mutually_exclusive, require_any
+from cloup.constraints import (
+    AnySet,
+    If,
+    accept_none,
+    mutually_exclusive,
+    require_any,
+)
 from pyhelper_utils.shell import run_command
 import pytest
 from rich.console import Console
@@ -37,6 +47,7 @@
 }
 LOGGER = get_logger(name="class_generator")
 TESTS_MANIFESTS_DIR = "class_generator/tests/manifests"
+OCP_RESOURCES_STR: str = "ocp_resources"
 
 
 def process_fields_args(
@@ -418,7 +429,7 @@ def generate_resource_file_from_dict(
     elif output_file:
         _output_file = output_file
     else:
-        _output_file = os.path.join("ocp_resources", f"{formatted_kind_str}.py")
+        _output_file = os.path.join(OCP_RESOURCES_STR, f"{formatted_kind_str}.py")
 
     if os.path.exists(_output_file):
         if overwrite:
@@ -481,7 +492,11 @@ def parse_explain(
         if debug_content:
             spec_out = debug_content.get("explain-spec", "")
     else:
-        rc, spec_out, _ = run_command(command=shlex.split(f"oc explain {kind}.spec"), check=False, log_errors=False)
+        rc, spec_out, _ = run_command(
+            command=shlex.split(f"oc explain {kind}.spec"),
+            check=False,
+            log_errors=False,
+        )
         if not rc:
             LOGGER.warning(f"{kind} spec not found, skipping")
 
@@ -532,7 +547,7 @@ def parse_explain(
             LOGGER.warning(
                 f"Missing API Group in Resource\n"
                 f"Please add `Resource.ApiGroup.{api_group_for_resource_api_group} = {api_group_real_name}` "
-                "manually into ocp_resources/resource.py under Resource class > ApiGroup class."
+                f"manually into {OCP_RESOURCES_STR}/resource.py under Resource class > ApiGroup class."
             )
 
     else:
@@ -543,7 +558,7 @@ def parse_explain(
             LOGGER.warning(
                 f"Missing API Version in Resource\n"
                 f"Please add `Resource.ApiVersion.{api_version_for_resource_api_version} = {resource_dict['VERSION']}` "
-                "manually into ocp_resources/resource.py under Resource class > ApiGroup class."
+                f"manually into {OCP_RESOURCES_STR}/resource.py under Resource class > ApiGroup class."
             )
 
     return resource_dict
@@ -693,6 +708,65 @@ def generate_class_generator_tests() -> None:
     )
 
 
+def delete_unchanged_files(updated_files: List[str]) -> List[str]:
+    for file_ in updated_files:
+        updated_file: str = file_.replace(".py", "_TEMP.py")
+        if os.path.exists(updated_file) and filecmp.cmp(file_, updated_file):
+            LOGGER.info(f"{file_} does not have any changes, deleting {updated_file} file.")
+            Path.unlink(Path(updated_file))
+            updated_files.remove(file_)
+
+    return updated_files
+
+
+def update_ocp_resources() -> None:
+    futures: List[Future] = []
+    updated_files: List[str] = []
+    exceptions: List[BaseException] = []
+
+    with ThreadPoolExecutor() as executor:
+        for obj in os.listdir(OCP_RESOURCES_STR):
+            filepath = os.path.join(OCP_RESOURCES_STR, obj)
+            if (
+                os.path.isfile(filepath)
+                and obj.endswith(".py")
+                and obj not in ("__init__.py", "resource.py", "utils.py")
+            ):
+                with open(filepath) as fd:
+                    data = fd.read()
+
+                if data.startswith("# Generated using"):
+                    kind = re.search(r"class\s+(.*?)\((Namespaced)?Resource", data)
+                    if kind:
+                        updated_files.append(filepath)
+                        futures.append(
+                            executor.submit(
+                                run_command,
+                                **{
+                                    "command": shlex.split(
+                                        f"poetry run python class_generator/class_generator.py --kind {kind.group(1)}"
+                                    ),
+                                    "log_errors": False,
+                                },
+                            )
+                        )
+
+        if futures:
+            for result in as_completed(futures):
+                if _exception := result.exception():
+                    exceptions.append(_exception)
+
+        updated_files_to_review = "\n".join(delete_unchanged_files(updated_files=updated_files))
+        LOGGER.warning(
+            f"The following files were updated:\n{updated_files_to_review}.\n"
+            "Please review the changes before commiting."
+        )
+
+        if exceptions:
+            exceptions_str = ", ".join(ex.args[1][-1] for ex in exceptions)
+            LOGGER.error(f"Failed to update resources:\n{exceptions_str}")
+
+
 @cloup.command("Resource class generator")
 @cloup.option("-i", "--interactive", is_flag=True, help="Enable interactive mode")
 @cloup.option(
@@ -731,12 +805,27 @@ def generate_class_generator_tests() -> None:
     is_flag=True,
     show_default=True,
 )
+@cloup.option(
+    "--update-resources",
+    help="Update all resources which generated by this script",
+    is_flag=True,
+    show_default=True,
+)
 @cloup.constraint(mutually_exclusive, ["add_tests", "debug"])
 @cloup.constraint(mutually_exclusive, ["add_tests", "output_file"])
 @cloup.constraint(mutually_exclusive, ["add_tests", "dry_run"])
+@cloup.constraint(mutually_exclusive, ["update_resources", "add_tests"])
+@cloup.constraint(mutually_exclusive, ["update_resources", "dry_run"])
+@cloup.constraint(mutually_exclusive, ["update_resources", "output_file"])
 @cloup.constraint(mutually_exclusive, ["interactive", "kind"])
-@cloup.constraint(If("debug_file", then=accept_none), ["interactive", "kind"])
-@cloup.constraint(require_any, ["interactive", "kind"])
+@cloup.constraint(
+    If(
+        AnySet("debug_file", "update_resources"),
+        then=accept_none,
+        else_=require_any,
+    ),
+    ["interactive", "kind"],
+)
 def main(
     kind: str,
     overwrite: bool,
@@ -747,9 +836,14 @@ def main(
     output_file: str,
     pdb: bool,
     add_tests: bool,
+    update_resources: bool,
 ):
     _ = pdb
 
+    if update_resources:
+        update_ocp_resources()
+        return
+
     class_generator(
         kind=kind,
         overwrite=overwrite,