feat: improved resources API

transpire/resources/__init__.py

@@ -3,5 +3,6 @@
 from .ingress import Ingress
 from .secret import Secret
 from .service import Service
+from .statefulset import StatefulSet
 
 __all__ = ["Deployment", "Ingress", "Service", "Secret", "ConfigMap"]

transpire/resources/base.py

@@ -12,6 +12,9 @@ class Resource(Generic[T]):
     def __init__(self):
         self.patches = []
 
+    def name(self) -> str:
+        return self.obj.metadata.name
+
     def patch(self, *fns: Callable[[dict], dict]) -> Self:
         self.patches.extend(fns)
         return self

transpire/resources/configmap.py

@@ -1,3 +1,4 @@
+from pathlib import Path
 from kubernetes import client
 
 from transpire.resources.base import Resource
@@ -19,3 +20,10 @@ def __init__(
             data=data,
         )
         super().__init__()
+
+    @staticmethod
+    def from_files(name, files: list[Path]):
+        data = {}
+        for file in files:
+            data[file.name] = file.read_text()
+        return ConfigMap(name=name, data=data)

transpire/resources/deployment.py

@@ -1,8 +1,9 @@
-from typing import List, Self, Union
+from typing import Any, List, Union
 
 from kubernetes import client
 
 from transpire.resources.base import Resource
+from transpire.resources.podspec import PodSpec
 
 
 class Deployment(Resource[client.V1Deployment]):
@@ -46,97 +47,11 @@ def __init__(
         )
         super().__init__()
 
-    def _init_env(self, container_id: int) -> client.V1Container:
-        container = self.obj.spec.template.spec.containers[container_id]
-
-        if container.env_from is None:
-            container.env_from = []
-        if container.env is None:
-            container.env = []
-
-        return container
-
-    def with_configmap_env(
-        self, name: str, *, mapping: dict[str, str] | None = None, container_id: int = 0
-    ) -> Self:
-        container = self._init_env(container_id)
-        if mapping is None:
-            container.env_from.append(
-                client.V1EnvFromSource(
-                    config_map_ref=client.V1ConfigMapEnvSource(
-                        name=name,
-                    )
-                )
-            )
-        else:
-            container.env.extend(
-                client.V1EnvVar(
-                    name=envvar_name,
-                    value_from=client.V1EnvVarSource(
-                        config_map_key_ref=client.V1ConfigMapKeySelector(
-                            name=name,
-                            key=cm_key,
-                        )
-                    ),
-                )
-                for envvar_name, cm_key in mapping.items()
-            )
-        return self
-
-    def with_secrets_env(
-        self, name: str, *, mapping: dict[str, str] | None = None, container_id: int = 0
-    ) -> Self:
-        container = self._init_env(container_id)
-        if mapping is None:
-            container.env_from.append(
-                client.V1EnvFromSource(
-                    secret_ref=client.V1SecretEnvSource(
-                        name=name,
-                    )
-                )
-            )
-        else:
-            container.env.extend(
-                client.V1EnvVar(
-                    name=envvar_name,
-                    value_from=client.V1EnvVarSource(
-                        secret_key_ref=client.V1SecretKeySelector(
-                            name=name,
-                            key=secret_key,
-                        )
-                    ),
-                )
-                for envvar_name, secret_key in mapping.items()
-            )
-        return self
-
-    def get_container(
-        self, name: str | None = None, *, remove: bool = False
-    ) -> client.V1Container:
-        container_list = self.obj.spec.template.spec.containers
-
-        if name is None:
-            if len(container_list) != 1:
-                raise ValueError("If multiple containers, must pass name.")
-            return container_list[0]
-
-        for i, container in enumerate(container_list):
-            if container.name == name:
-                if remove:
-                    return container_list.pop(i)
-                return container
-
-        raise ValueError(f"No such container: {name}")
-
-    def add_container(self, container: client.V1Container) -> int:
-        container_list = self.obj.spec.template.spec.containers
-
-        names = set(c.name for c in container_list)
-        if container.name in names:
-            raise ValueError(f"Can't use name {container.name}, already in use.")
-
-        container_list.append(container)
-        return len(container_list) - 1
+    def pod_spec(self) -> PodSpec:
+        return PodSpec(self.obj.spec.template.spec)
 
     def get_selector(self) -> dict[str, str]:
         return {self.SELECTOR_LABEL: self.obj.metadata.name}
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self.pod_spec(), name)