Fix pyright type errors across codebase

- Add pyrightconfig.json for pyright 1.1.408 TOML parsing workaround - Add NoReturn annotations to fatal() functions for proper type narrowing - Add None checks and assertions after require=True get_record() calls - Fix AttrDict class with __getattr__ for dynamic attribute access - Add type annotations and casts for Kubernetes client objects - Store compose config as DockerDeployer instance attributes - Filter None values from dotenv and environment mappings - Use hasattr/getattr patterns for optional container attributes Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-22 01:10:36 -05:00
parent cd3d908d0d
commit dd856af2d3
29 changed files with 512 additions and 267 deletions
@@ -17,7 +17,7 @@ import os
 import base64

 from kubernetes import client
-from typing import Any, List, Set
+from typing import Any, List, Optional, Set

 from stack_orchestrator.opts import opts
 from stack_orchestrator.util import env_var_map_from_file
@@ -51,7 +51,7 @@ DEFAULT_CONTAINER_RESOURCES = Resources(


 def to_k8s_resource_requirements(resources: Resources) -> client.V1ResourceRequirements:
-    def to_dict(limits: ResourceLimits):
+    def to_dict(limits: Optional[ResourceLimits]):
        if not limits:
            return None

@@ -83,9 +83,11 @@ class ClusterInfo:
        self.parsed_pod_yaml_map = parsed_pod_files_map_from_file_names(pod_files)
        # Find the set of images in the pods
        self.image_set = images_for_deployment(pod_files)
-        self.environment_variables = DeployEnvVars(
-            env_var_map_from_file(compose_env_file)
-        )
+        # Filter out None values from env file
+        env_vars = {
+            k: v for k, v in env_var_map_from_file(compose_env_file).items() if v
+        }
+        self.environment_variables = DeployEnvVars(env_vars)
        self.app_name = deployment_name
        self.spec = spec
        if opts.o.debug:
@@ -214,6 +216,7 @@ class ClusterInfo:

    # TODO: suppoprt multiple services
    def get_service(self):
+        port = None
        for pod_name in self.parsed_pod_yaml_map:
            pod = self.parsed_pod_yaml_map[pod_name]
            services = pod["services"]
@@ -223,6 +226,8 @@ class ClusterInfo:
                    port = int(service_info["ports"][0])
                    if opts.o.debug:
                        print(f"service port: {port}")
+        if port is None:
+            return None
        service = client.V1Service(
            metadata=client.V1ObjectMeta(name=f"{self.app_name}-service"),
            spec=client.V1ServiceSpec(
@@ -287,9 +292,9 @@ class ClusterInfo:
                    print(f"{cfg_map_name} not in pod files")
                continue

-            if not cfg_map_path.startswith("/"):
+            if not cfg_map_path.startswith("/") and self.spec.file_path is not None:
                cfg_map_path = os.path.join(
-                    os.path.dirname(self.spec.file_path), cfg_map_path
+                    os.path.dirname(str(self.spec.file_path)), cfg_map_path
                )

            # Read in all the files at a single-level of the directory.
@@ -367,8 +372,9 @@ class ClusterInfo:
        return result

    # TODO: put things like image pull policy into an object-scope struct
-    def get_deployment(self, image_pull_policy: str = None):
+    def get_deployment(self, image_pull_policy: Optional[str] = None):
        containers = []
+        services = {}
        resources = self.spec.get_container_resources()
        if not resources:
            resources = DEFAULT_CONTAINER_RESOURCES
@@ -16,7 +16,8 @@ from datetime import datetime, timezone

 from pathlib import Path
 from kubernetes import client, config
-from typing import List
+from kubernetes.client.exceptions import ApiException
+from typing import Any, Dict, List, Optional, cast

 from stack_orchestrator import constants
 from stack_orchestrator.deploy.deployer import Deployer, DeployerConfigGenerator
@@ -50,7 +51,7 @@ class AttrDict(dict):
        self.__dict__ = self


-def _check_delete_exception(e: client.exceptions.ApiException):
+def _check_delete_exception(e: ApiException) -> None:
    if e.status == 404:
        if opts.o.debug:
            print("Failed to delete object, continuing")
@@ -189,18 +190,25 @@ class K8sDeployer(Deployer):
        if opts.o.debug:
            print(f"Sending this deployment: {deployment}")
        if not opts.o.dry_run:
-            deployment_resp = self.apps_api.create_namespaced_deployment(
-                body=deployment, namespace=self.k8s_namespace
+            deployment_resp = cast(
+                client.V1Deployment,
+                self.apps_api.create_namespaced_deployment(
+                    body=deployment, namespace=self.k8s_namespace
+                ),
            )
            if opts.o.debug:
                print("Deployment created:")
-                ns = deployment_resp.metadata.namespace
-                name = deployment_resp.metadata.name
-                gen = deployment_resp.metadata.generation
-                img = deployment_resp.spec.template.spec.containers[0].image
-                print(f"{ns} {name} {gen} {img}")
+                meta = deployment_resp.metadata
+                spec = deployment_resp.spec
+                if meta and spec and spec.template.spec:
+                    ns = meta.namespace
+                    name = meta.name
+                    gen = meta.generation
+                    containers = spec.template.spec.containers
+                    img = containers[0].image if containers else None
+                    print(f"{ns} {name} {gen} {img}")

-        service: client.V1Service = self.cluster_info.get_service()
+        service = self.cluster_info.get_service()
        if opts.o.debug:
            print(f"Sending this service: {service}")
        if not opts.o.dry_run:
@@ -254,7 +262,7 @@ class K8sDeployer(Deployer):
                # Create the kind cluster
                create_cluster(
                    self.kind_cluster_name,
-                    self.deployment_dir.joinpath(constants.kind_config_filename),
+                    str(self.deployment_dir.joinpath(constants.kind_config_filename)),
                )
                # Ensure the referenced containers are copied into kind
                load_images_into_kind(
@@ -286,7 +294,7 @@ class K8sDeployer(Deployer):
            if certificate:
                print(f"Using existing certificate: {certificate}")

-        ingress: client.V1Ingress = self.cluster_info.get_ingress(
+        ingress = self.cluster_info.get_ingress(
            use_tls=use_tls, certificate=certificate
        )
        if ingress:
@@ -333,7 +341,7 @@ class K8sDeployer(Deployer):
                    if opts.o.debug:
                        print("PV deleted:")
                        print(f"{pv_resp}")
-                except client.exceptions.ApiException as e:
+                except ApiException as e:
                    _check_delete_exception(e)

            # Figure out the PVCs for this deployment
@@ -348,7 +356,7 @@ class K8sDeployer(Deployer):
                    if opts.o.debug:
                        print("PVCs deleted:")
                        print(f"{pvc_resp}")
-                except client.exceptions.ApiException as e:
+                except ApiException as e:
                    _check_delete_exception(e)

        # Figure out the ConfigMaps for this deployment
@@ -363,40 +371,40 @@ class K8sDeployer(Deployer):
                if opts.o.debug:
                    print("ConfigMap deleted:")
                    print(f"{cfg_map_resp}")
-            except client.exceptions.ApiException as e:
+            except ApiException as e:
                _check_delete_exception(e)

        deployment = self.cluster_info.get_deployment()
        if opts.o.debug:
            print(f"Deleting this deployment: {deployment}")
-        try:
-            self.apps_api.delete_namespaced_deployment(
-                name=deployment.metadata.name, namespace=self.k8s_namespace
-            )
-        except client.exceptions.ApiException as e:
-            _check_delete_exception(e)
+        if deployment and deployment.metadata and deployment.metadata.name:
+            try:
+                self.apps_api.delete_namespaced_deployment(
+                    name=deployment.metadata.name, namespace=self.k8s_namespace
+                )
+            except ApiException as e:
+                _check_delete_exception(e)

-        service: client.V1Service = self.cluster_info.get_service()
+        service = self.cluster_info.get_service()
        if opts.o.debug:
            print(f"Deleting service: {service}")
-        try:
-            self.core_api.delete_namespaced_service(
-                namespace=self.k8s_namespace, name=service.metadata.name
-            )
-        except client.exceptions.ApiException as e:
-            _check_delete_exception(e)
+        if service and service.metadata and service.metadata.name:
+            try:
+                self.core_api.delete_namespaced_service(
+                    namespace=self.k8s_namespace, name=service.metadata.name
+                )
+            except ApiException as e:
+                _check_delete_exception(e)

-        ingress: client.V1Ingress = self.cluster_info.get_ingress(
-            use_tls=not self.is_kind()
-        )
-        if ingress:
+        ingress = self.cluster_info.get_ingress(use_tls=not self.is_kind())
+        if ingress and ingress.metadata and ingress.metadata.name:
            if opts.o.debug:
                print(f"Deleting this ingress: {ingress}")
            try:
                self.networking_api.delete_namespaced_ingress(
                    name=ingress.metadata.name, namespace=self.k8s_namespace
                )
-            except client.exceptions.ApiException as e:
+            except ApiException as e:
                _check_delete_exception(e)
        else:
            if opts.o.debug:
@@ -406,12 +414,13 @@ class K8sDeployer(Deployer):
        for nodeport in nodeports:
            if opts.o.debug:
                print(f"Deleting this nodeport: {nodeport}")
-            try:
-                self.core_api.delete_namespaced_service(
-                    namespace=self.k8s_namespace, name=nodeport.metadata.name
-                )
-            except client.exceptions.ApiException as e:
-                _check_delete_exception(e)
+            if nodeport.metadata and nodeport.metadata.name:
+                try:
+                    self.core_api.delete_namespaced_service(
+                        namespace=self.k8s_namespace, name=nodeport.metadata.name
+                    )
+                except ApiException as e:
+                    _check_delete_exception(e)
        else:
            if opts.o.debug:
                print("No nodeport to delete")
@@ -428,8 +437,9 @@ class K8sDeployer(Deployer):

        if all_pods.items:
            for p in all_pods.items:
-                if f"{self.cluster_info.app_name}-deployment" in p.metadata.name:
-                    pods.append(p)
+                if p.metadata and p.metadata.name:
+                    if f"{self.cluster_info.app_name}-deployment" in p.metadata.name:
+                        pods.append(p)

        if not pods:
            return
@@ -438,24 +448,39 @@ class K8sDeployer(Deployer):
        ip = "?"
        tls = "?"
        try:
-            ingress = self.networking_api.read_namespaced_ingress(
-                namespace=self.k8s_namespace,
-                name=self.cluster_info.get_ingress().metadata.name,
+            cluster_ingress = self.cluster_info.get_ingress()
+            if cluster_ingress is None or cluster_ingress.metadata is None:
+                return
+            ingress = cast(
+                client.V1Ingress,
+                self.networking_api.read_namespaced_ingress(
+                    namespace=self.k8s_namespace,
+                    name=cluster_ingress.metadata.name,
+                ),
            )
+            if not ingress.spec or not ingress.spec.tls or not ingress.spec.rules:
+                return

-            cert = self.custom_obj_api.get_namespaced_custom_object(
-                group="cert-manager.io",
-                version="v1",
-                namespace=self.k8s_namespace,
-                plural="certificates",
-                name=ingress.spec.tls[0].secret_name,
+            cert = cast(
+                Dict[str, Any],
+                self.custom_obj_api.get_namespaced_custom_object(
+                    group="cert-manager.io",
+                    version="v1",
+                    namespace=self.k8s_namespace,
+                    plural="certificates",
+                    name=ingress.spec.tls[0].secret_name,
+                ),
            )

            hostname = ingress.spec.rules[0].host
-            ip = ingress.status.load_balancer.ingress[0].ip
+            if ingress.status and ingress.status.load_balancer:
+                lb_ingress = ingress.status.load_balancer.ingress
+                if lb_ingress:
+                    ip = lb_ingress[0].ip or "?"
+            cert_status = cert.get("status", {})
            tls = "notBefore: %s; notAfter: %s; names: %s" % (
-                cert["status"]["notBefore"],
-                cert["status"]["notAfter"],
+                cert_status.get("notBefore", "?"),
+                cert_status.get("notAfter", "?"),
                ingress.spec.tls[0].hosts,
            )
        except:  # noqa: E722
@@ -469,6 +494,8 @@ class K8sDeployer(Deployer):
        print("Pods:")

        for p in pods:
+            if not p.metadata:
+                continue
            ns = p.metadata.namespace
            name = p.metadata.name
            if p.metadata.deletion_timestamp:
@@ -539,7 +566,7 @@ class K8sDeployer(Deployer):
                    container_log_lines = container_log.splitlines()
                    for line in container_log_lines:
                        log_data += f"{container}: {line}\n"
-            except client.exceptions.ApiException as e:
+            except ApiException as e:
                if opts.o.debug:
                    print(f"Error from read_namespaced_pod_log: {e}")
                log_data = "******* No logs available ********\n"
@@ -548,25 +575,44 @@ class K8sDeployer(Deployer):
    def update(self):
        self.connect_api()
        ref_deployment = self.cluster_info.get_deployment()
+        if not ref_deployment or not ref_deployment.metadata:
+            return
+        ref_name = ref_deployment.metadata.name
+        if not ref_name:
+            return

-        deployment = self.apps_api.read_namespaced_deployment(
-            name=ref_deployment.metadata.name, namespace=self.k8s_namespace
+        deployment = cast(
+            client.V1Deployment,
+            self.apps_api.read_namespaced_deployment(
+                name=ref_name, namespace=self.k8s_namespace
+            ),
        )
+        if not deployment.spec or not deployment.spec.template:
+            return
+        template_spec = deployment.spec.template.spec
+        if not template_spec or not template_spec.containers:
+            return

-        new_env = ref_deployment.spec.template.spec.containers[0].env
-        for container in deployment.spec.template.spec.containers:
-            old_env = container.env
-            if old_env != new_env:
-                container.env = new_env
+        ref_spec = ref_deployment.spec
+        if ref_spec and ref_spec.template and ref_spec.template.spec:
+            ref_containers = ref_spec.template.spec.containers
+            if ref_containers:
+                new_env = ref_containers[0].env
+                for container in template_spec.containers:
+                    old_env = container.env
+                    if old_env != new_env:
+                        container.env = new_env

-        deployment.spec.template.metadata.annotations = {
-            "kubectl.kubernetes.io/restartedAt": datetime.utcnow()
-            .replace(tzinfo=timezone.utc)
-            .isoformat()
-        }
+        template_meta = deployment.spec.template.metadata
+        if template_meta:
+            template_meta.annotations = {
+                "kubectl.kubernetes.io/restartedAt": datetime.utcnow()
+                .replace(tzinfo=timezone.utc)
+                .isoformat()
+            }

        self.apps_api.patch_namespaced_deployment(
-            name=ref_deployment.metadata.name,
+            name=ref_name,
            namespace=self.k8s_namespace,
            body=deployment,
        )
@@ -585,7 +631,7 @@ class K8sDeployer(Deployer):
        # We need to figure out how to do this -- check why we're being called first
        pass

-    def run_job(self, job_name: str, helm_release: str = None):
+    def run_job(self, job_name: str, helm_release: Optional[str] = None):
        if not opts.o.dry_run:
            from stack_orchestrator.deploy.k8s.helm.job_runner import run_helm_job

@@ -138,6 +138,8 @@ def generate_helm_chart(
    """

    parsed_stack = get_parsed_stack_config(stack_path)
+    if parsed_stack is None:
+        error_exit(f"Failed to parse stack config: {stack_path}")
    stack_name = parsed_stack.get("name", stack_path)

    # 1. Check Kompose availability
@@ -185,22 +187,28 @@ def generate_helm_chart(
    compose_files = []
    for pod in pods:
        pod_file = get_pod_file_path(stack_path, parsed_stack, pod)
-        if not pod_file.exists():
-            error_exit(f"Pod file not found: {pod_file}")
-        compose_files.append(pod_file)
+        if pod_file is None:
+            error_exit(f"Pod file path not found for pod: {pod}")
+        pod_file_path = Path(pod_file) if isinstance(pod_file, str) else pod_file
+        if not pod_file_path.exists():
+            error_exit(f"Pod file not found: {pod_file_path}")
+        compose_files.append(pod_file_path)
        if opts.o.debug:
-            print(f"Found compose file: {pod_file.name}")
+            print(f"Found compose file: {pod_file_path.name}")

    # Add job compose files
    job_files = []
    for job in jobs:
        job_file = get_job_file_path(stack_path, parsed_stack, job)
-        if not job_file.exists():
-            error_exit(f"Job file not found: {job_file}")
-        compose_files.append(job_file)
-        job_files.append(job_file)
+        if job_file is None:
+            error_exit(f"Job file path not found for job: {job}")
+        job_file_path = Path(job_file) if isinstance(job_file, str) else job_file
+        if not job_file_path.exists():
+            error_exit(f"Job file not found: {job_file_path}")
+        compose_files.append(job_file_path)
+        job_files.append(job_file_path)
        if opts.o.debug:
-            print(f"Found job compose file: {job_file.name}")
+            print(f"Found job compose file: {job_file_path.name}")

    try:
        version = get_kompose_version()
@@ -18,6 +18,7 @@ import tempfile
 import os
 import json
 from pathlib import Path
+from typing import Optional
 from stack_orchestrator.util import get_yaml


@@ -50,7 +51,7 @@ def get_release_name_from_chart(chart_dir: Path) -> str:
 def run_helm_job(
    chart_dir: Path,
    job_name: str,
-    release: str = None,
+    release: Optional[str] = None,
    namespace: str = "default",
    timeout: int = 600,
    verbose: bool = False,
@@ -16,7 +16,7 @@
 import subprocess
 import shutil
 from pathlib import Path
-from typing import List
+from typing import List, Optional


 def check_kompose_available() -> bool:
@@ -53,7 +53,7 @@ def get_kompose_version() -> str:


 def convert_to_helm_chart(
-    compose_files: List[Path], output_dir: Path, chart_name: str = None
+    compose_files: List[Path], output_dir: Path, chart_name: Optional[str] = None
 ) -> str:
    """
    Invoke kompose to convert Docker Compose files to a Helm chart.
@@ -18,7 +18,7 @@ import os
 from pathlib import Path
 import subprocess
 import re
-from typing import Set, Mapping, List
+from typing import Set, Mapping, List, Optional, cast

 from stack_orchestrator.util import get_k8s_dir, error_exit
 from stack_orchestrator.opts import opts
@@ -75,8 +75,10 @@ def wait_for_ingress_in_kind():
            label_selector="app.kubernetes.io/component=controller",
            timeout_seconds=30,
        ):
-            if event["object"].status.container_statuses:
-                if event["object"].status.container_statuses[0].ready is True:
+            event_dict = cast(dict, event)
+            pod = cast(client.V1Pod, event_dict.get("object"))
+            if pod and pod.status and pod.status.container_statuses:
+                if pod.status.container_statuses[0].ready is True:
                    if warned_waiting:
                        print("Ingress controller is ready")
                    return
@@ -119,14 +121,18 @@ def pods_in_deployment(core_api: client.CoreV1Api, deployment_name: str):
    return pods


-def containers_in_pod(core_api: client.CoreV1Api, pod_name: str):
-    containers = []
-    pod_response = core_api.read_namespaced_pod(pod_name, namespace="default")
+def containers_in_pod(core_api: client.CoreV1Api, pod_name: str) -> List[str]:
+    containers: List[str] = []
+    pod_response = cast(
+        client.V1Pod, core_api.read_namespaced_pod(pod_name, namespace="default")
+    )
    if opts.o.debug:
        print(f"pod_response: {pod_response}")
-    pod_containers = pod_response.spec.containers
-    for pod_container in pod_containers:
-        containers.append(pod_container.name)
+    if not pod_response.spec or not pod_response.spec.containers:
+        return containers
+    for pod_container in pod_response.spec.containers:
+        if pod_container.name:
+            containers.append(pod_container.name)
    return containers


@@ -351,7 +357,9 @@ def merge_envs(a: Mapping[str, str], b: Mapping[str, str]) -> Mapping[str, str]:
    return result


-def _expand_shell_vars(raw_val: str, env_map: Mapping[str, str] = None) -> str:
+def _expand_shell_vars(
+    raw_val: str, env_map: Optional[Mapping[str, str]] = None
+) -> str:
    # Expand docker-compose style variable substitution:
    # ${VAR} - use VAR value or empty string
    # ${VAR:-default} - use VAR value or default if unset/empty
@@ -376,7 +384,7 @@ def _expand_shell_vars(raw_val: str, env_map: Mapping[str, str] = None) -> str:


 def envs_from_compose_file(
-    compose_file_envs: Mapping[str, str], env_map: Mapping[str, str] = None
+    compose_file_envs: Mapping[str, str], env_map: Optional[Mapping[str, str]] = None
 ) -> Mapping[str, str]:
    result = {}
    for env_var, env_val in compose_file_envs.items():