diff --git a/pulumi/__main__.py b/pulumi/__main__.py index 235a764..d62b6ed 100644 --- a/pulumi/__main__.py +++ b/pulumi/__main__.py @@ -1,5 +1,4 @@ # ./pulumi/__main__.py -# Description: from typing import List, Dict, Any diff --git a/pulumi/core/README.md b/pulumi/core/README.md index 80a8c12..a53a49f 100644 --- a/pulumi/core/README.md +++ b/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 diff --git a/pulumi/core/init.py b/pulumi/core/init.py index 404d180..567f7f2 100644 --- a/pulumi/core/init.py +++ b/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 ( diff --git a/pulumi/core/metadata.py b/pulumi/core/metadata.py index 6b260f8..32bb7d4 100644 --- a/pulumi/core/metadata.py +++ b/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", ""), diff --git a/pulumi/core/utils.py b/pulumi/core/utils.py index 99a9b85..8a0f813 100644 --- a/pulumi/core/utils.py +++ b/pulumi/core/utils.py @@ -6,45 +6,52 @@ 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) @@ -52,6 +59,15 @@ def sanitize_label_value(value: str) -> str: 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) diff --git a/pulumi/core/versions.py b/pulumi/core/versions.py index 6c25b57..86b7500 100644 --- a/pulumi/core/versions.py +++ b/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