Jelajahi Sumber

refactor: remove all schema infrastructure from module classes

- Remove schema imports from all 6 modules (compose, terraform, ansible, kubernetes, helm, packer)
- Remove _load_json_spec_as_dict() functions from all modules
- Remove _SchemaDict classes from all modules
- Remove SCHEMAS initialization and spec variables
- Remove schema_version and schemas attributes from module classes
- Update module docstrings from 'with multi-schema support' to basic module descriptions
- Remove OrderedDict imports no longer needed

All templates are now fully self-contained with explicit variable definitions in template.yaml files. No schema inheritance or merging occurs at runtime.
xcad 6 bulan lalu
induk
melakukan
2863004251

+ 6 - 6
cli/core/template/template.py

@@ -383,19 +383,19 @@ class Template:
 
     def _merge_specs(self) -> dict:
         """Process template specs into merged format.
-        
+
         Since schemas are no longer used, this just uses the template specs directly
         and warns about unused variables.
         """
         if not self.template_specs:
             return {}
-        
+
         # Warn about unused variables in spec
         self._warn_about_unused_variables(self.template_specs)
-        
+
         # Create VariableCollection from template specs
         collection = VariableCollection(self.template_specs)
-        
+
         # Convert back to dict format
         merged_spec = {}
         for section_key, section in collection.get_sections().items():
@@ -820,10 +820,10 @@ class Template:
         if self.__variables is None:
             # Process template specs (merge and warn about unused variables)
             merged_specs = self._merge_specs()
-            
+
             # Validate that all used variables are defined
             self._validate_variable_definitions(self.used_variables, merged_specs)
-            
+
             # Filter specs to only used variables
             filtered_specs = self._filter_specs_to_used(
                 self.used_variables,

+ 1 - 70
cli/modules/ansible/__init__.py

@@ -1,88 +1,19 @@
-"""Ansible module with multi-schema support."""
+"""Ansible module."""
 
 import logging
-from collections import OrderedDict
 
 from ...core.module import Module
 from ...core.registry import registry
-from ...core.schema import has_schema, list_versions, load_schema
 
 logger = logging.getLogger(__name__)
 
 
-def _load_json_spec_as_dict(version: str) -> OrderedDict:
-    """Load JSON schema and convert to dict format for backward compatibility.
-
-    Args:
-        version: Schema version
-
-    Returns:
-        OrderedDict in the same format as Python specs
-    """
-    logger.debug(f"Loading ansible schema {version} from JSON")
-    json_spec = load_schema("ansible", version)
-
-    # Convert JSON array format to OrderedDict format
-    spec_dict = OrderedDict()
-    for section_data in json_spec:
-        section_key = section_data["key"]
-
-        # Build section dict
-        section_dict = {}
-        if "title" in section_data:
-            section_dict["title"] = section_data["title"]
-        if "description" in section_data:
-            section_dict["description"] = section_data["description"]
-        if "toggle" in section_data:
-            section_dict["toggle"] = section_data["toggle"]
-        if "required" in section_data:
-            section_dict["required"] = section_data["required"]
-        if "needs" in section_data:
-            section_dict["needs"] = section_data["needs"]
-
-        # Convert vars array to dict
-        vars_dict = OrderedDict()
-        for var_data in section_data["vars"]:
-            var_name = var_data["name"]
-            var_dict = {k: v for k, v in var_data.items() if k != "name"}
-            vars_dict[var_name] = var_dict
-
-        section_dict["vars"] = vars_dict
-        spec_dict[section_key] = section_dict
-
-    return spec_dict
-
-
-# Schema version mapping - loads JSON schemas on-demand
-class _SchemaDict(dict):
-    """Dict subclass that loads JSON schemas on-demand."""
-
-    def __getitem__(self, version):
-        if not has_schema("ansible", version):
-            raise KeyError(
-                f"Schema version {version} not found for ansible module. "
-                f"Available: {', '.join(list_versions('ansible'))}"
-            )
-        return _load_json_spec_as_dict(version)
-
-    def __contains__(self, version):
-        return has_schema("ansible", version)
-
-
-# Initialize schema dict
-SCHEMAS = _SchemaDict()
-
-# Default spec - load latest version
-spec = _load_json_spec_as_dict("1.0")
-
 
 class AnsibleModule(Module):
     """Ansible module."""
 
     name = "ansible"
     description = "Manage Ansible configurations"
-    schema_version = "1.0"  # Current schema version supported by this module
-    schemas = SCHEMAS  # Available schema versions
 
 
 registry.register(AnsibleModule)

+ 2 - 72
cli/modules/compose/__init__.py

@@ -1,93 +1,23 @@
-"""Docker Compose module with multi-schema support."""
+"""Docker Compose module."""
 
 import logging
-from collections import OrderedDict
-from typing import Annotated
 
 from typer import Argument, Option
+from typing import Annotated
 
 from ...core.module import Module
 from ...core.module.base_commands import validate_templates
 from ...core.registry import registry
-from ...core.schema import has_schema, list_versions, load_schema
 from .validate import run_docker_validation
 
 logger = logging.getLogger(__name__)
 
 
-def _load_json_spec_as_dict(version: str) -> OrderedDict:
-    """Load JSON schema and convert to dict format for backward compatibility.
-
-    Args:
-        version: Schema version
-
-    Returns:
-        OrderedDict in the same format as Python specs
-    """
-    logger.debug(f"Loading compose schema {version} from JSON")
-    json_spec = load_schema("compose", version)
-
-    # Convert JSON array format to OrderedDict format
-    spec_dict = OrderedDict()
-    for section_data in json_spec:
-        section_key = section_data["key"]
-
-        # Build section dict
-        section_dict = {}
-        if "title" in section_data:
-            section_dict["title"] = section_data["title"]
-        if "description" in section_data:
-            section_dict["description"] = section_data["description"]
-        if "toggle" in section_data:
-            section_dict["toggle"] = section_data["toggle"]
-        if "required" in section_data:
-            section_dict["required"] = section_data["required"]
-        if "needs" in section_data:
-            section_dict["needs"] = section_data["needs"]
-
-        # Convert vars array to dict
-        vars_dict = OrderedDict()
-        for var_data in section_data["vars"]:
-            var_name = var_data["name"]
-            var_dict = {k: v for k, v in var_data.items() if k != "name"}
-            vars_dict[var_name] = var_dict
-
-        section_dict["vars"] = vars_dict
-        spec_dict[section_key] = section_dict
-
-    return spec_dict
-
-
-# Schema version mapping - loads JSON schemas on-demand
-class _SchemaDict(dict):
-    """Dict subclass that loads JSON schemas on-demand."""
-
-    def __getitem__(self, version):
-        if not has_schema("compose", version):
-            raise KeyError(
-                f"Schema version {version} not found for compose module. "
-                f"Available: {', '.join(list_versions('compose'))}"
-            )
-        return _load_json_spec_as_dict(version)
-
-    def __contains__(self, version):
-        return has_schema("compose", version)
-
-
-# Initialize schema dict
-SCHEMAS = _SchemaDict()
-
-# Default spec - load latest version
-spec = _load_json_spec_as_dict("1.2")
-
-
 class ComposeModule(Module):
     """Docker Compose module with extended validation."""
 
     name = "compose"
     description = "Manage Docker Compose configurations"
-    schema_version = "1.2"  # Current schema version supported by this module
-    schemas = SCHEMAS  # Available schema versions
 
     def validate(  # noqa: PLR0913
         self,

+ 1 - 69
cli/modules/helm/__init__.py

@@ -1,87 +1,19 @@
-"""Helm module with multi-schema support."""
+"""Helm module."""
 
 import logging
-from collections import OrderedDict
 
 from ...core.module import Module
 from ...core.registry import registry
-from ...core.schema import has_schema, list_versions, load_schema
 
 logger = logging.getLogger(__name__)
 
 
-def _load_json_spec_as_dict(version: str) -> OrderedDict:
-    """Load JSON schema and convert to dict format for backward compatibility.
-
-    Args:
-        version: Schema version
-
-    Returns:
-        OrderedDict in the same format as Python specs
-    """
-    logger.debug(f"Loading helm schema {version} from JSON")
-    json_spec = load_schema("helm", version)
-
-    # Convert JSON array format to OrderedDict format
-    spec_dict = OrderedDict()
-    for section_data in json_spec:
-        section_key = section_data["key"]
-
-        # Build section dict
-        section_dict = {}
-        if "title" in section_data:
-            section_dict["title"] = section_data["title"]
-        if "description" in section_data:
-            section_dict["description"] = section_data["description"]
-        if "toggle" in section_data:
-            section_dict["toggle"] = section_data["toggle"]
-        if "required" in section_data:
-            section_dict["required"] = section_data["required"]
-        if "needs" in section_data:
-            section_dict["needs"] = section_data["needs"]
-
-        # Convert vars array to dict
-        vars_dict = OrderedDict()
-        for var_data in section_data["vars"]:
-            var_name = var_data["name"]
-            var_dict = {k: v for k, v in var_data.items() if k != "name"}
-            vars_dict[var_name] = var_dict
-
-        section_dict["vars"] = vars_dict
-        spec_dict[section_key] = section_dict
-
-    return spec_dict
-
-
-# Schema version mapping - loads JSON schemas on-demand
-class _SchemaDict(dict):
-    """Dict subclass that loads JSON schemas on-demand."""
-
-    def __getitem__(self, version):
-        if not has_schema("helm", version):
-            raise KeyError(
-                f"Schema version {version} not found for helm module. Available: {', '.join(list_versions('helm'))}"
-            )
-        return _load_json_spec_as_dict(version)
-
-    def __contains__(self, version):
-        return has_schema("helm", version)
-
-
-# Initialize schema dict
-SCHEMAS = _SchemaDict()
-
-# Default spec - load latest version
-spec = _load_json_spec_as_dict("1.0")
-
 
 class HelmModule(Module):
     """Helm module."""
 
     name = "helm"
     description = "Manage Helm configurations"
-    schema_version = "1.0"  # Current schema version supported by this module
-    schemas = SCHEMAS  # Available schema versions
 
 
 registry.register(HelmModule)

+ 1 - 70
cli/modules/kubernetes/__init__.py

@@ -1,88 +1,19 @@
-"""Kubernetes module with multi-schema support."""
+"""Kubernetes module."""
 
 import logging
-from collections import OrderedDict
 
 from ...core.module import Module
 from ...core.registry import registry
-from ...core.schema import has_schema, list_versions, load_schema
 
 logger = logging.getLogger(__name__)
 
 
-def _load_json_spec_as_dict(version: str) -> OrderedDict:
-    """Load JSON schema and convert to dict format for backward compatibility.
-
-    Args:
-        version: Schema version
-
-    Returns:
-        OrderedDict in the same format as Python specs
-    """
-    logger.debug(f"Loading kubernetes schema {version} from JSON")
-    json_spec = load_schema("kubernetes", version)
-
-    # Convert JSON array format to OrderedDict format
-    spec_dict = OrderedDict()
-    for section_data in json_spec:
-        section_key = section_data["key"]
-
-        # Build section dict
-        section_dict = {}
-        if "title" in section_data:
-            section_dict["title"] = section_data["title"]
-        if "description" in section_data:
-            section_dict["description"] = section_data["description"]
-        if "toggle" in section_data:
-            section_dict["toggle"] = section_data["toggle"]
-        if "required" in section_data:
-            section_dict["required"] = section_data["required"]
-        if "needs" in section_data:
-            section_dict["needs"] = section_data["needs"]
-
-        # Convert vars array to dict
-        vars_dict = OrderedDict()
-        for var_data in section_data["vars"]:
-            var_name = var_data["name"]
-            var_dict = {k: v for k, v in var_data.items() if k != "name"}
-            vars_dict[var_name] = var_dict
-
-        section_dict["vars"] = vars_dict
-        spec_dict[section_key] = section_dict
-
-    return spec_dict
-
-
-# Schema version mapping - loads JSON schemas on-demand
-class _SchemaDict(dict):
-    """Dict subclass that loads JSON schemas on-demand."""
-
-    def __getitem__(self, version):
-        if not has_schema("kubernetes", version):
-            raise KeyError(
-                f"Schema version {version} not found for kubernetes module. "
-                f"Available: {', '.join(list_versions('kubernetes'))}"
-            )
-        return _load_json_spec_as_dict(version)
-
-    def __contains__(self, version):
-        return has_schema("kubernetes", version)
-
-
-# Initialize schema dict
-SCHEMAS = _SchemaDict()
-
-# Default spec - load latest version
-spec = _load_json_spec_as_dict("1.0")
-
 
 class KubernetesModule(Module):
     """Kubernetes module."""
 
     name = "kubernetes"
     description = "Manage Kubernetes configurations"
-    schema_version = "1.0"  # Current schema version supported by this module
-    schemas = SCHEMAS  # Available schema versions
 
 
 registry.register(KubernetesModule)

+ 1 - 69
cli/modules/packer/__init__.py

@@ -1,87 +1,19 @@
-"""Packer module with multi-schema support."""
+"""Packer module."""
 
 import logging
-from collections import OrderedDict
 
 from ...core.module import Module
 from ...core.registry import registry
-from ...core.schema import has_schema, list_versions, load_schema
 
 logger = logging.getLogger(__name__)
 
 
-def _load_json_spec_as_dict(version: str) -> OrderedDict:
-    """Load JSON schema and convert to dict format for backward compatibility.
-
-    Args:
-        version: Schema version
-
-    Returns:
-        OrderedDict in the same format as Python specs
-    """
-    logger.debug(f"Loading packer schema {version} from JSON")
-    json_spec = load_schema("packer", version)
-
-    # Convert JSON array format to OrderedDict format
-    spec_dict = OrderedDict()
-    for section_data in json_spec:
-        section_key = section_data["key"]
-
-        # Build section dict
-        section_dict = {}
-        if "title" in section_data:
-            section_dict["title"] = section_data["title"]
-        if "description" in section_data:
-            section_dict["description"] = section_data["description"]
-        if "toggle" in section_data:
-            section_dict["toggle"] = section_data["toggle"]
-        if "required" in section_data:
-            section_dict["required"] = section_data["required"]
-        if "needs" in section_data:
-            section_dict["needs"] = section_data["needs"]
-
-        # Convert vars array to dict
-        vars_dict = OrderedDict()
-        for var_data in section_data["vars"]:
-            var_name = var_data["name"]
-            var_dict = {k: v for k, v in var_data.items() if k != "name"}
-            vars_dict[var_name] = var_dict
-
-        section_dict["vars"] = vars_dict
-        spec_dict[section_key] = section_dict
-
-    return spec_dict
-
-
-# Schema version mapping - loads JSON schemas on-demand
-class _SchemaDict(dict):
-    """Dict subclass that loads JSON schemas on-demand."""
-
-    def __getitem__(self, version):
-        if not has_schema("packer", version):
-            raise KeyError(
-                f"Schema version {version} not found for packer module. Available: {', '.join(list_versions('packer'))}"
-            )
-        return _load_json_spec_as_dict(version)
-
-    def __contains__(self, version):
-        return has_schema("packer", version)
-
-
-# Initialize schema dict
-SCHEMAS = _SchemaDict()
-
-# Default spec - load latest version
-spec = _load_json_spec_as_dict("1.0")
-
 
 class PackerModule(Module):
     """Packer module."""
 
     name = "packer"
     description = "Manage Packer configurations"
-    schema_version = "1.0"  # Current schema version supported by this module
-    schemas = SCHEMAS  # Available schema versions
 
 
 registry.register(PackerModule)

+ 1 - 70
cli/modules/terraform/__init__.py

@@ -1,88 +1,19 @@
-"""Terraform module with multi-schema support."""
+"""Terraform module."""
 
 import logging
-from collections import OrderedDict
 
 from ...core.module import Module
 from ...core.registry import registry
-from ...core.schema import has_schema, list_versions, load_schema
 
 logger = logging.getLogger(__name__)
 
 
-def _load_json_spec_as_dict(version: str) -> OrderedDict:
-    """Load JSON schema and convert to dict format for backward compatibility.
-
-    Args:
-        version: Schema version
-
-    Returns:
-        OrderedDict in the same format as Python specs
-    """
-    logger.debug(f"Loading terraform schema {version} from JSON")
-    json_spec = load_schema("terraform", version)
-
-    # Convert JSON array format to OrderedDict format
-    spec_dict = OrderedDict()
-    for section_data in json_spec:
-        section_key = section_data["key"]
-
-        # Build section dict
-        section_dict = {}
-        if "title" in section_data:
-            section_dict["title"] = section_data["title"]
-        if "description" in section_data:
-            section_dict["description"] = section_data["description"]
-        if "toggle" in section_data:
-            section_dict["toggle"] = section_data["toggle"]
-        if "required" in section_data:
-            section_dict["required"] = section_data["required"]
-        if "needs" in section_data:
-            section_dict["needs"] = section_data["needs"]
-
-        # Convert vars array to dict
-        vars_dict = OrderedDict()
-        for var_data in section_data["vars"]:
-            var_name = var_data["name"]
-            var_dict = {k: v for k, v in var_data.items() if k != "name"}
-            vars_dict[var_name] = var_dict
-
-        section_dict["vars"] = vars_dict
-        spec_dict[section_key] = section_dict
-
-    return spec_dict
-
-
-# Schema version mapping - loads JSON schemas on-demand
-class _SchemaDict(dict):
-    """Dict subclass that loads JSON schemas on-demand."""
-
-    def __getitem__(self, version):
-        if not has_schema("terraform", version):
-            raise KeyError(
-                f"Schema version {version} not found for terraform module. "
-                f"Available: {', '.join(list_versions('terraform'))}"
-            )
-        return _load_json_spec_as_dict(version)
-
-    def __contains__(self, version):
-        return has_schema("terraform", version)
-
-
-# Initialize schema dict
-SCHEMAS = _SchemaDict()
-
-# Default spec - load latest version
-spec = _load_json_spec_as_dict("1.0")
-
 
 class TerraformModule(Module):
     """Terraform module."""
 
     name = "terraform"
     description = "Manage Terraform configurations"
-    schema_version = "1.0"  # Current schema version supported by this module
-    schemas = SCHEMAS  # Available schema versions
 
 
 registry.register(TerraformModule)