refine core and module codebase

pulumi/__main__.py

@@ -10,7 +10,6 @@
 from core.config import export_results
 from core.deploy_module import deploy_module
 
-
 def main():
     try:
         init = initialize_pulumi()
@@ -33,7 +32,6 @@ def main():
         pulumi.log.error(f"Deployment failed: {str(e)}")
         raise
 
-
 def deploy_modules(
         modules: List[str],
         config: pulumi.Config,
@@ -56,6 +54,5 @@ def deploy_modules(
             configurations=configurations,
         )
 
-
 if __name__ == "__main__":
     main()

pulumi/core/config.py

@@ -13,10 +13,10 @@
 import pulumi
 
 def get_module_config(
-    module_name: str,
-    config: pulumi.Config,
-    default_versions: Dict[str, Any],
-) -> Tuple[Dict[str, Any], bool]:
+        module_name: str,
+        config: pulumi.Config,
+        default_versions: Dict[str, Any],
+    ) -> Tuple[Dict[str, Any], bool]:
     """
     Retrieves and prepares the configuration for a module.
 
@@ -33,7 +33,11 @@ def get_module_config(
     module_config['version'] = module_config.get('version', default_versions.get(module_name))
     return module_config, module_enabled
 
-def export_results(versions: Dict[str, str], configurations: Dict[str, Dict[str, Any]], compliance: Dict[str, Any]):
+def export_results(
+        versions: Dict[str, str],
+        configurations: Dict[str, Dict[str, Any]],
+        compliance: Dict[str, Any]
+    ):
     """
     Exports the results of the deployment processes including versions, configurations, and compliance information.
 

pulumi/core/deploy_module.py

@@ -4,7 +4,7 @@
 import inspect
 import pulumi
 import pulumi_kubernetes as k8s
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Type
 from .introspection import discover_config_class, discover_deploy_function
 from .config import get_module_config
 
@@ -33,7 +33,6 @@ def deploy_module(
         TypeError: If any arguments have incorrect types.
         ValueError: If any module-specific errors occur.
     """
-    # Validate parameter types
     if not isinstance(module_name, str):
         raise TypeError("module_name must be a string")
     if not isinstance(config, pulumi.Config):
@@ -49,29 +48,24 @@ def deploy_module(
     if not isinstance(configurations, dict):
         raise TypeError("configurations must be a dictionary")
 
-    # Get the module's configuration from Pulumi config
     module_config_dict, module_enabled = get_module_config(module_name, config, default_versions)
 
     if module_enabled:
-        # Discover the configuration class and deploy function dynamically
         ModuleConfigClass = discover_config_class(module_name)
         deploy_func = discover_deploy_function(module_name)
 
         config_obj = ModuleConfigClass.merge(module_config_dict)
 
-        # Infer the required argument name for the deploy function
         deploy_func_args = inspect.signature(deploy_func).parameters.keys()
-        config_arg_name = list(deploy_func_args)[0]  # Assuming the first argument is the config object
+        config_arg_name = list(deploy_func_args)[0]
 
-        # Deploy the module using its deploy function with the correct arguments
         try:
             result = deploy_func(
                 **{config_arg_name: config_obj},
                 global_depends_on=global_depends_on,
                 k8s_provider=k8s_provider,
             )
 
-            # Handle the result based on its structure
             if isinstance(result, tuple) and len(result) == 3:
                 version, release, exported_value = result
             elif isinstance(result, tuple) and len(result) == 2:
@@ -80,20 +74,16 @@ def deploy_module(
             else:
                 raise ValueError(f"Unexpected return value structure from {module_name} deploy function")
 
-            # Record the deployed version and configuration
             versions[module_name] = version
             configurations[module_name] = {"enabled": module_enabled}
 
-            # Export any additional values if needed
             if exported_value:
                 pulumi.export(f"{module_name}_exported_value", exported_value)
 
-            # Add the release to global dependencies to maintain resource ordering
             global_depends_on.append(release)
 
         except Exception as e:
             pulumi.log.error(f"Deployment failed for module {module_name}: {str(e)}")
             raise
-
     else:
         pulumi.log.info(f"Module {module_name} is not enabled.")

pulumi/core/helm_chart_versions.py

@@ -14,38 +14,18 @@
 CHART_NOT_FOUND = "Chart not found"
 
 def is_stable_version(version_str: str) -> bool:
-    """
-    Check if the version string is a valid and stable semantic version.
-
-    Args:
-        version_str (str): The version string to check.
-
-    Returns:
-        bool: True if the version is stable, False otherwise.
-    """
     try:
         parsed_version = parse_version(version_str)
         return isinstance(parsed_version, Version) and not parsed_version.is_prerelease and not parsed_version.is_devrelease
     except InvalidVersion:
         return False
 
 def get_latest_helm_chart_version(url: str, chart_name: str) -> str:
-    """
-    Fetches the latest stable version of a Helm chart from a given URL.
-
-    Args:
-        url (str): The URL of the Helm chart repository.
-        chart_name (str): The name of the Helm chart.
-
-    Returns:
-        str: The latest stable version of the Helm chart, or an error message if the chart is not found or an error occurs during fetching.
-    """
     try:
         logging.info(f"Fetching URL: {url}")
         response = requests.get(url)
         response.raise_for_status()
 
-        # Parse the YAML content
         index = yaml.safe_load(response.content)
         if chart_name in index['entries']:
             chart_versions = index['entries'][chart_name]

pulumi/core/init.py

@@ -20,66 +20,64 @@
 from .utils import generate_global_transformations
 
 def initialize_pulumi() -> Dict[str, Any]:
-    """
-    Initializes the Pulumi configuration, Kubernetes provider, and global resources.
-
-    Returns:
-        Dict[str, Any]: A dictionary containing initialized resources.
-    """
     config = pulumi.Config()
     stack_name = pulumi.get_stack()
     project_name = pulumi.get_project()
 
-    default_versions = load_default_versions(config)
-    versions: Dict[str, str] = {}
-    configurations: Dict[str, Dict[str, Any]] = {}
-    global_depends_on: List[pulumi.Resource] = []
+    try:
+        default_versions = load_default_versions(config)
+        versions: Dict[str, str] = {}
+        configurations: Dict[str, Dict[str, Any]] = {}
+        global_depends_on: List[pulumi.Resource] = []
 
-    kubernetes_config = config.get_object("kubernetes") or {}
-    kubeconfig = kubernetes_config.get("kubeconfig") or os.getenv('KUBECONFIG')
-    kubernetes_context = kubernetes_config.get("context")
+        kubernetes_config = config.get_object("kubernetes") or {}
+        kubeconfig = kubernetes_config.get("kubeconfig") or os.getenv('KUBECONFIG')
+        kubernetes_context = kubernetes_config.get("context")
 
-    pulumi.log.info(f"Kubeconfig: {kubeconfig}")
-    pulumi.log.info(f"Kubernetes context: {kubernetes_context}")
+        pulumi.log.info(f"Kubeconfig: {kubeconfig}")
+        pulumi.log.info(f"Kubernetes context: {kubernetes_context}")
 
-    k8s_provider = Provider(
-        "k8sProvider",
-        kubeconfig=kubeconfig,
-        context=kubernetes_context,
-    )
+        k8s_provider = Provider(
+            "k8sProvider",
+            kubeconfig=kubeconfig,
+            context=kubernetes_context,
+        )
 
-    git_info = collect_git_info()
-    configurations["source_repository"] = {
-        "remote": git_info["remote"],
-        "branch": git_info["branch"],
-        "commit": git_info["commit"]
-    }
+        git_info = collect_git_info()
+        configurations["source_repository"] = {
+            "remote": git_info["remote"],
+            "branch": git_info["branch"],
+            "commit": git_info["commit"]
+        }
 
-    compliance_config_dict = config.get_object('compliance') or {}
-    compliance_config = ComplianceConfig.merge(compliance_config_dict)
-    compliance_labels = generate_compliance_labels(compliance_config)
-    compliance_annotations = generate_compliance_annotations(compliance_config)
+        compliance_config_dict = config.get_object('compliance') or {}
+        compliance_config = ComplianceConfig.merge(compliance_config_dict)
+        compliance_labels = generate_compliance_labels(compliance_config)
+        compliance_annotations = generate_compliance_annotations(compliance_config)
 
-    git_labels = generate_git_labels(git_info)
-    git_annotations = generate_git_annotations(git_info)
-    global_labels = {**compliance_labels, **git_labels}
-    global_annotations = {**compliance_annotations, **git_annotations}
+        git_labels = generate_git_labels(git_info)
+        git_annotations = generate_git_annotations(git_info)
+        global_labels = {**compliance_labels, **git_labels}
+        global_annotations = {**compliance_annotations, **git_annotations}
 
-    set_global_labels(global_labels)
-    set_global_annotations(global_annotations)
-    generate_global_transformations(global_labels, global_annotations)
+        set_global_labels(global_labels)
+        set_global_annotations(global_annotations)
+        generate_global_transformations(global_labels, global_annotations)
 
-    return {
-        "config": config,
-        "stack_name": stack_name,
-        "project_name": project_name,
-        "default_versions": default_versions,
-        "versions": versions,
-        "configurations": configurations,
-        "global_depends_on": global_depends_on,
-        "k8s_provider": k8s_provider,
-        "git_info": git_info,
-        "compliance_config": compliance_config,
-        "global_labels": global_labels,
-        "global_annotations": global_annotations,
-    }
+        return {
+            "config": config,
+            "stack_name": stack_name,
+            "project_name": project_name,
+            "default_versions": default_versions,
+            "versions": versions,
+            "configurations": configurations,
+            "global_depends_on": global_depends_on,
+            "k8s_provider": k8s_provider,
+            "git_info": git_info,
+            "compliance_config": compliance_config,
+            "global_labels": global_labels,
+            "global_annotations": global_annotations,
+        }
+    except Exception as e:
+        pulumi.log.error(f"Initialization error: {str(e)}")
+        raise

pulumi/core/introspection.py

@@ -3,7 +3,7 @@
 
 import inspect
 import importlib
-from typing import Type, Callable, Optional
+from typing import Type, Callable
 
 def discover_config_class(module_name: str) -> Type:
     """