fix:(class-generator) improve generation reliability and code quality (#2493)

Author: myakove

.flake8

@@ -9,70 +9,3 @@ exclude =
     Pipfile.*,
     docs/*,
     .cache/*
-
-fcn_exclude_functions =
-    update,
-    getLogger,
-    loads,
-    get,
-    next,
-    setdefault,
-    dumps,
-    request,
-    list,
-    warning,
-    join,
-    rmtree,
-    open,
-    os,
-    write,
-    append,
-    sorted,
-    compile,
-    int,
-    extend,
-    sleep,
-    check_output,
-    sub_resource_level,
-    func,
-    LOGGER,
-    findall,
-    raises,
-    contextlib,
-    re,
-    requests,
-    signal,
-    yaml,
-    benedict,
-    logger,
-    warn,
-    pytest,
-    json,
-    template,
-    meta,
-    tmp_path_factory,
-    tmpdir_factory,
-    Path,
-    writelines,
-    submit,
-    click,
-    ast,
-    filecmp,
-    datetime,
-    subprocess,
-    urlparse,
-    parse_qs,
-    self,
-    defaultdict,
-    add_row,
-    get_args,
-    get_type_hints,
-    shutil,
-    asyncio,
-    __class__,
-    cls,
-    validate,
-    relative_to,
-
-enable-extensions =
-    FCN,

.gitignore

@@ -141,3 +141,6 @@ CLAUDE.md
 
 # Backup directories
 .backups/
+
+# Large schema files (we keep the compressed versions)
+class_generator/schema/__resources-mappings.json

class_generator/cli.py

@@ -16,7 +16,7 @@
 from class_generator.core.coverage import analyze_coverage, generate_report
 from class_generator.core.discovery import discover_generated_resources
 from class_generator.core.generator import class_generator
-from class_generator.core.schema import update_kind_schema
+from class_generator.core.schema import update_kind_schema, ClusterVersionError
 from class_generator.tests.test_generation import generate_class_generator_tests
 from ocp_resources.utils.utils import convert_camel_case_to_snake_case
 
@@ -53,7 +53,11 @@ def handle_schema_update(update_schema: bool, generate_missing: bool) -> bool:
     """
     if update_schema:
         LOGGER.info("Updating resource schema...")
-        update_kind_schema()
+        try:
+            update_kind_schema()
+        except (RuntimeError, IOError, ClusterVersionError) as e:
+            LOGGER.error(f"Failed to update schema: {e}")
+            sys.exit(1)
 
         # If only updating schema (not generating), exit
         if not generate_missing:
@@ -317,13 +321,17 @@ def handle_normal_kind_generation(
             # Create backup if file exists
             create_backup_if_needed(target_file=target_file, backup_dir=backup_dir)
 
-        class_generator(
-            kind=kind,
-            overwrite=overwrite,
-            dry_run=dry_run,
-            output_file=output_file,
-            add_tests=add_tests,
-        )
+        try:
+            class_generator(
+                kind=kind,
+                overwrite=overwrite,
+                dry_run=dry_run,
+                output_file=output_file,
+                add_tests=add_tests,
+            )
+        except Exception as e:
+            LOGGER.error(f"Failed to generate {kind}: {e}")
+            sys.exit(1)
 
         if backup_dir and not dry_run:
             LOGGER.info(f"Backup files stored in: {backup_dir}")
@@ -339,13 +347,17 @@ def generate_with_backup(kind_to_generate: str) -> list[str]:
                 # Create backup if file exists
                 create_backup_if_needed(target_file=target_file, backup_dir=backup_dir)
 
-            return class_generator(
-                kind=kind_to_generate,
-                overwrite=overwrite,
-                dry_run=dry_run,
-                output_file=output_file,
-                add_tests=add_tests,
-            )
+            try:
+                return class_generator(
+                    kind=kind_to_generate,
+                    overwrite=overwrite,
+                    dry_run=dry_run,
+                    output_file=output_file,
+                    add_tests=add_tests,
+                )
+            except Exception as e:
+                LOGGER.error(f"Failed to generate {kind_to_generate}: {e}")
+                return []
 
         futures: list[Future] = []
         with ThreadPoolExecutor(max_workers=10) as executor:

class_generator/constants.py

@@ -1,4 +1,4 @@
-"""Constants used throughout the class generator."""
+"""Constants used throughout the class generator and schema utilities."""
 
 import keyword
 from pathlib import Path
@@ -11,6 +11,8 @@
 TESTS_MANIFESTS_DIR: Path = Path("class_generator") / "tests" / "manifests"
 SCHEMA_DIR: Path = Path("class_generator") / "schema"
 RESOURCES_MAPPING_FILE: Path = SCHEMA_DIR / "__resources-mappings.json"
+RESOURCES_MAPPING_ARCHIVE: Path = SCHEMA_DIR / "__resources-mappings.json.gz"
+DEFINITIONS_FILE: Path = SCHEMA_DIR / "_definitions.json"
 
 # Description constants
 MISSING_DESCRIPTION_STR: str = "No field description from API"

class_generator/core/generator.py

@@ -2,12 +2,9 @@
 
 import filecmp
 import os
-import shlex
-import sys
 from pathlib import Path
 from typing import Any
 
-from pyhelper_utils.shell import run_command
 from rich.console import Console
 from rich.syntax import Syntax
 from simple_logger.logger import get_logger
@@ -86,9 +83,10 @@ def generate_resource_file_from_dict(
             LOGGER.warning(f"Overwriting {_output_file}")
 
         else:
-            temp_output_file = _output_file.replace(".py", "_TEMP.py")
-            LOGGER.warning(f"{_output_file} already exists, using {temp_output_file}")
-            _output_file = temp_output_file
+            if not dry_run:
+                temp_output_file = _output_file.replace(".py", "_TEMP.py")
+                LOGGER.warning(f"{_output_file} already exists, using {temp_output_file}")
+                _output_file = temp_output_file
 
     if _user_code.strip() or _user_imports.strip():
         output += f"{_user_imports}{rendered}{_user_code}"
@@ -114,6 +112,7 @@ def class_generator(
     add_tests: bool = False,
     called_from_cli: bool = True,
     update_schema_executed: bool = False,
+    called_from_test: bool = False,
 ) -> list[str]:
     """
     Generates a class for a given Kind.
@@ -134,55 +133,50 @@ def class_generator(
     LOGGER.info(f"Generating class for {kind}")
     kind = kind.lower()
     kind_and_namespaced_mappings = read_resources_mapping_file().get(kind)
-    if not kind_and_namespaced_mappings:
+    if not called_from_test and not kind_and_namespaced_mappings:
         if called_from_cli:
             if update_schema_executed:
-                LOGGER.error(f"{kind} not found in {RESOURCES_MAPPING_FILE} after update-schema executed.")
-                sys.exit(1)
-
-            # Use a loop to handle retries instead of recursion
-            max_retries = 3
-            retry_count = 0
-
-            while retry_count < max_retries:
-                # Validate user input with a loop
-                while True:
-                    run_update_schema = (
-                        input(
-                            f"{kind} not found in {RESOURCES_MAPPING_FILE}, Do you want to run --update-schema and retry? [Y/N]: "
-                        )
-                        .strip()
-                        .lower()
-                    )
-
-                    if run_update_schema in ["y", "n"]:
-                        break
-                    else:
-                        print("Invalid input. Please enter 'Y' or 'N'.")
-
-                if run_update_schema == "n":
-                    sys.exit(1)
-
-                # User chose 'y' - update schema and retry
-                LOGGER.info(f"Updating schema (attempt {retry_count + 1}/{max_retries})...")
-                update_kind_schema()
+                error_msg = f"{kind} not found in {RESOURCES_MAPPING_FILE} after update-schema executed."
+                LOGGER.error(error_msg)
+                raise RuntimeError(error_msg)
+
+            run_update_schema = (
+                input(
+                    f"{kind} not found in {RESOURCES_MAPPING_FILE}, Do you want to run --update-schema and retry? [Y/N]: "
+                )
+                .strip()
+                .lower()
+            )
 
-                # Re-read the mapping file to check if kind is now available
-                kind_and_namespaced_mappings = read_resources_mapping_file().get(kind)
-                if kind_and_namespaced_mappings:
-                    # Kind found, break out of retry loop to continue processing
-                    break
-                else:
-                    retry_count += 1
-                    if retry_count < max_retries:
-                        LOGGER.warning(
-                            f"{kind} still not found after schema update. Retry {retry_count}/{max_retries}."
-                        )
-                    else:
-                        LOGGER.error(
-                            f"{kind} not found in {RESOURCES_MAPPING_FILE} after {max_retries} update-schema attempts."
-                        )
-                        sys.exit(1)
+            if run_update_schema != "y":
+                raise RuntimeError(f"User declined to update schema for {kind}")
+
+            # User chose 'y' - update schema and retry
+            LOGGER.info("Updating schema")
+            try:
+                update_kind_schema()
+            except (RuntimeError, IOError) as e:
+                error_msg = f"Failed to update schema: {e}"
+                LOGGER.error(error_msg)
+                raise RuntimeError(error_msg) from e
+
+            # Re-read the mapping file to check if kind is now available
+            kind_and_namespaced_mappings = read_resources_mapping_file(skip_cache=True).get(kind)
+            if kind_and_namespaced_mappings:
+                return class_generator(
+                    kind=kind,
+                    overwrite=overwrite,
+                    dry_run=dry_run,
+                    output_file=output_file,
+                    output_dir=output_dir,
+                    add_tests=add_tests,
+                    called_from_cli=called_from_cli,
+                    update_schema_executed=True,
+                )
+            else:
+                error_msg = f"{kind} not found in {RESOURCES_MAPPING_FILE} after update-schema attempt."
+                LOGGER.error(error_msg)
+                raise RuntimeError(error_msg)
 
         else:
             LOGGER.error(f"{kind} not found in {RESOURCES_MAPPING_FILE}, Please run --update-schema.")
@@ -215,29 +209,7 @@ def class_generator(
             output_dir=output_dir,
         )
 
-        if not dry_run:
-            try:
-                rc, stdout, stderr = run_command(
-                    command=shlex.split(f"uvx pre-commit run --files {generated_py_file}"),
-                    verify_stderr=False,
-                    check=False,
-                )
-                # Check if the command failed
-                if not rc:
-                    LOGGER.warning(
-                        f"Pre-commit hooks failed for {generated_py_file}. "
-                        f"This is non-fatal and generation will continue."
-                    )
-                    if stderr:
-                        LOGGER.debug(f"Pre-commit stderr: {stderr}")
-                    if stdout:
-                        LOGGER.debug(f"Pre-commit stdout: {stdout}")
-            except Exception as e:
-                LOGGER.error(
-                    f"Failed to run pre-commit hooks for {generated_py_file}: {e}. "
-                    f"This is non-fatal and generation will continue."
-                )
-
+        if not dry_run and not called_from_test:
             if orig_filename != generated_py_file and filecmp.cmp(orig_filename, generated_py_file):
                 LOGGER.warning(f"File {orig_filename} was not updated, deleting {generated_py_file}")
                 Path.unlink(Path(generated_py_file))