tidy core

pulumi/__main__.py

@@ -1,5 +1,4 @@
 # ./pulumi/__main__.py
-# Description:
 
 from typing import List, Dict, Any
 

pulumi/core/README.md

@@ -45,7 +45,7 @@ Located in `pulumi/core/compliance.py`, this section deals with generating compl
 
 - **`ComplianceConfig`**
   Central configuration object for compliance settings.
-- **`FismaConfig`, `NistConfig`, `ScipConfig`**
+- **`FismaConfig`, `NistConfig`, and `ScipConfig`**
   Nested configuration types used within `ComplianceConfig`.
 
 ### Configuration
@@ -137,7 +137,7 @@ Located in `pulumi/core/types.py`, this section defines shared configuration typ
 
 - **`NamespaceConfig`**
   Configuration object for Kubernetes namespaces.
-- **`ComplianceConfig`, `FismaConfig`, `NistConfig`, `ScipConfig`**
+- **`ComplianceConfig`, `FismaConfig`, `NistConfig`, and `ScipConfig`**
   Configuration objects for compliance settings.
 
 ### Utilities

pulumi/core/init.py

@@ -1,11 +1,10 @@
 # ./pulumi/core/init.py
 # Description: Initializes Pulumi configuration, Kubernetes provider, and global resources.
-
 import os
 import pulumi
 import pulumi_kubernetes as k8s
 from pulumi_kubernetes import Provider
-from typing import Dict, List, Any
+from typing import Dict, Any, List
 
 from .versions import load_default_versions
 from .metadata import (

pulumi/core/metadata.py

@@ -22,18 +22,48 @@ def __new__(cls, *args, **kwargs):
         return cls._instance
 
 def set_global_labels(labels: Dict[str, str]):
+    """
+    Sets global labels.
+
+    Args:
+        labels (Dict[str, str]): The global labels.
+    """
     MetadataSingleton()._data["_global_labels"] = labels
 
 def set_global_annotations(annotations: Dict[str, str]):
+    """
+    Sets global annotations.
+
+    Args:
+        annotations (Dict[str, str]): The global annotations.
+    """
     MetadataSingleton()._data["_global_annotations"] = annotations
 
 def get_global_labels() -> Dict[str, str]:
+    """
+    Retrieves global labels.
+
+    Returns:
+        Dict[str, str]: The global labels.
+    """
     return MetadataSingleton()._data["_global_labels"]
 
 def get_global_annotations() -> Dict[str, str]:
+    """
+    Retrieves global annotations.
+
+    Returns:
+        Dict[str, str]: The global annotations.
+    """
     return MetadataSingleton()._data["_global_annotations"]
 
 def collect_git_info() -> Dict[str, str]:
+    """
+    Collects Git repository information.
+
+    Returns:
+        Dict[str, str]: The Git information.
+    """
     try:
         remote = subprocess.check_output(['git', 'config', '--get', 'remote.origin.url'], stderr=subprocess.STDOUT).strip().decode('utf-8')
         branch = subprocess.check_output(['git', 'rev-parse', '--abbrev-ref', 'HEAD'], stderr=subprocess.STDOUT).strip().decode('utf-8')
@@ -44,12 +74,30 @@ def collect_git_info() -> Dict[str, str]:
         return {'remote': 'N/A', 'branch': 'N/A', 'commit': 'N/A'}
 
 def generate_git_labels(git_info: Dict[str, str]) -> Dict[str, str]:
+    """
+    Generates git-related labels.
+
+    Args:
+        git_info (Dict[str, str]): The Git information.
+
+    Returns:
+        Dict[str, str]: The git-related labels.
+    """
     return {
         "git.branch": git_info.get("branch", ""),
         "git.commit": git_info.get("commit", "")[:7],
     }
 
 def generate_git_annotations(git_info: Dict[str, str]) -> Dict[str, str]:
+    """
+    Generates git-related annotations.
+
+    Args:
+        git_info (Dict[str, str]): The Git information.
+
+    Returns:
+        Dict[str, str]: The git-related annotations.
+    """
     return {
         "git.remote": git_info.get("remote", ""),
         "git.commit.full": git_info.get("commit", ""),

pulumi/core/utils.py

@@ -6,52 +6,68 @@
 import pulumi_kubernetes as k8s
 from typing import Optional, Dict, Any
 
-
 def set_resource_metadata(metadata: Any, global_labels: Dict[str, str], global_annotations: Dict[str, str]):
+    """
+    Updates resource metadata with global labels and annotations.
+
+    Args:
+        metadata (Any): Metadata to update.
+        global_labels (Dict[str, str]): Global labels to apply.
+        global_annotations (Dict[str, str]): Global annotations to apply.
+    """
     if isinstance(metadata, dict):
-        if metadata.get('labels') is None:
-            metadata['labels'] = {}
         metadata.setdefault('labels', {}).update(global_labels)
-
-        if metadata.get('annotations') is None:
-            metadata['annotations'] = {}
         metadata.setdefault('annotations', {}).update(global_annotations)
-
     elif isinstance(metadata, k8s.meta.v1.ObjectMetaArgs):
         if metadata.labels is None:
             metadata.labels = {}
         metadata.labels.update(global_labels)
-
         if metadata.annotations is None:
             metadata.annotations = {}
         metadata.annotations.update(global_annotations)
 
-
 def generate_global_transformations(global_labels: Dict[str, str], global_annotations: Dict[str, str]):
+    """
+    Generates global transformations for resources.
+
+    Args:
+        global_labels (Dict[str, str]): Global labels to apply.
+        global_annotations (Dict[str, str]): Global annotations to apply.
+    """
     def global_transform(args: pulumi.ResourceTransformationArgs) -> Optional[pulumi.ResourceTransformationResult]:
         props = args.props
-        if 'metadata' in props:
-            metadata = props['metadata']
-            set_resource_metadata(metadata, global_labels, global_annotations)
-            props['metadata'] = metadata
-        else:
-            props['metadata'] = {
-                'labels': global_labels,
-                'annotations': global_annotations
-            }
+        props.setdefault('metadata', {})
+        set_resource_metadata(props['metadata'], global_labels, global_annotations)
         return pulumi.ResourceTransformationResult(props, args.opts)
 
     pulumi.runtime.register_stack_transformation(global_transform)
 
-
 def sanitize_label_value(value: str) -> str:
+    """
+    Sanitizes a label value to comply with Kubernetes naming conventions.
+
+    Args:
+        value (str): The value to sanitize.
+
+    Returns:
+        str: The sanitized value.
+    """
     value = value.lower()
     sanitized = re.sub(r'[^a-z0-9_.-]', '-', value)
     sanitized = re.sub(r'^[^a-z0-9]+', '', sanitized)
     sanitized = re.sub(r'[^a-z0-9]+$', '', sanitized)
     return sanitized[:63]
 
 def extract_repo_name(remote_url: str) -> str:
+    """
+    Extracts the repository name from a Git remote URL.
+
+    Args:
+        remote_url (str): The Git remote URL.
+
+    Returns:
+        str: The repository name.
+    """
     match = re.search(r'[:/]([^/:]+/[^/\.]+)(\.git)?$', remote_url)
     if match:
         return match.group(1)

pulumi/core/versions.py

@@ -3,6 +3,7 @@
 import os
 import pulumi
 import requests
+from typing import Dict
 
 #DEFAULT_VERSIONS_URL_TEMPLATE = 'https://github.com/containercraft/kargo/releases/latest/download/'
 # TODO: replace with official github releases artifact URLs when released