More mypy fixes (#412)

kr8s/_auth.py

@@ -6,6 +6,7 @@
 import os
 import pathlib
 import ssl
+from typing import Union
 
 import anyio
 
@@ -24,30 +25,30 @@ def __init__(
         namespace=None,
         context=None,
     ) -> None:
-        self.server = None
-        self.client_cert_file = None
-        self.client_key_file = None
-        self.server_ca_file = None
-        self.token = None
-        self.namespace = namespace
-        self.active_context = None
+        self.server: str = None
+        self.client_cert_file: Union[str, pathlib.Path] = None
+        self.client_key_file: Union[str, pathlib.Path] = None
+        self.server_ca_file: Union[str, pathlib.Path] = None
+        self.token: str = None
+        self.namespace: str = namespace
+        self.active_context: str = None
         self.kubeconfig: KubeConfigSet = None
-        self.tls_server_name = None
-        self._url = url
-        self._insecure_skip_tls_verify = False
-        self._use_context = context
-        self._context = None
-        self._cluster = None
-        self._user = None
-        self._serviceaccount = (
+        self.tls_server_name: str = None
+        self._url: str = url
+        self._insecure_skip_tls_verify: bool = False
+        self._use_context: str = context
+        self._context: dict = None
+        self._cluster: dict = None
+        self._user: dict = None
+        self._serviceaccount: str = (
             serviceaccount
             if serviceaccount is not None
             else "/var/run/secrets/kubernetes.io/serviceaccount"
         )
-        self._kubeconfig_path_or_dict = kubeconfig or os.environ.get(
-            "KUBECONFIG", "~/.kube/config"
+        self._kubeconfig_path_or_dict: Union[dict, str, pathlib.Path] = (
+            kubeconfig or os.environ.get("KUBECONFIG", "~/.kube/config")
         )
-        self.__auth_lock = anyio.Lock()
+        self.__auth_lock: anyio.Lock = anyio.Lock()
 
     def __await__(self):
         async def f():

kr8s/_data_utils.py

@@ -65,7 +65,7 @@ def dot_to_nested_dict(dot_notated_key: str, value: Any) -> Dict:
         dot notated key.
     """
     keys = dot_notated_key.split(".")
-    nested_dict = {}
+    nested_dict: dict = {}
     for key in reversed(keys):
         if not nested_dict:
             nested_dict[key] = value

kr8s/_objects.py

@@ -54,10 +54,15 @@
 class APIObject:
     """Base class for Kubernetes objects."""
 
-    namespaced = False
-    scalable = False
-    scalable_spec = "replicas"
-    _asyncio = True
+    version: str
+    endpoint: str
+    kind: str
+    plural: str
+    singular: str
+    namespaced: bool = False
+    scalable: bool = False
+    scalable_spec: str = "replicas"
+    _asyncio: bool = True
 
     def __init__(
         self, resource: dict, namespace: Optional[str] = None, api: Optional[Api] = None
@@ -124,7 +129,7 @@ def api(self, value):
         self._api = value
 
     @property
-    def raw(self) -> str:
+    def raw(self) -> Any:
         """Raw object returned from the Kubernetes API."""
         self._raw.update({"kind": self.kind, "apiVersion": self.version})
         return self._raw
@@ -862,7 +867,7 @@ async def logs(
             ...     print(line)
 
         """
-        params = {}
+        params: dict[Any, Any] = {}
         if follow:
             params["follow"] = "true"
         if container is not None:
@@ -941,7 +946,7 @@ async def async_exec(
         command: List[str],
         *,
         container: Optional[str] = None,
-        stdin: Optional[Union(str | bytes | BinaryIO)] = None,
+        stdin: Optional[Union[str | bytes | BinaryIO]] = None,
         stdout: Optional[BinaryIO] = None,
         stderr: Optional[BinaryIO] = None,
         check: bool = True,
@@ -969,7 +974,7 @@ async def exec(
         command: List[str],
         *,
         container: Optional[str] = None,
-        stdin: Optional[Union(str | bytes | BinaryIO)] = None,
+        stdin: Optional[Union[str | bytes | BinaryIO]] = None,
         stdout: Optional[BinaryIO] = None,
         stderr: Optional[BinaryIO] = None,
         check: bool = True,

kr8s/_testutils.py

@@ -5,7 +5,7 @@
 
 
 @contextlib.contextmanager
-def set_env(**environ) -> None:
+def set_env(**environ):
     """
     Temporarily set the process environment variables.
 

kr8s/conftest.py

@@ -119,7 +119,7 @@ def run_id():
 
 
 @pytest.fixture(autouse=True)
-def ns(k8s_cluster, run_id) -> str:
+def ns(k8s_cluster, run_id):
     name = f"kr8s-pytest-{run_id}-{uuid.uuid4().hex[:4]}"
     k8s_cluster.kubectl("create", "namespace", name)
     yield name

kr8s/tests/test_auth.py

@@ -110,7 +110,7 @@ async def test_kubeconfig_multi_paths_same(k8s_cluster):
 
 
 async def test_kubeconfig_multi_paths_diff(k8s_cluster, tmp_path):
-    kubeconfig1: Path = k8s_cluster.kubeconfig_path
+    kubeconfig1 = k8s_cluster.kubeconfig_path
     kubeconfig2 = Path(tmp_path / "kubeconfig").write_bytes(kubeconfig1.read_bytes())
     kubeconfig_multi_str = f"{kubeconfig1}:{kubeconfig2}"
     api = await kr8s.asyncio.api(kubeconfig=kubeconfig_multi_str)

pyproject.toml

@@ -28,6 +28,7 @@ test = [
     "pytest-rerunfailures>=11.1.2",
     "pytest-cov>=4.0.0",
     "trio>=0.22.0",
+    "types-pyyaml>=6.0",
     "lightkube>=0.13.0",
     "kubernetes>=26.1.0",
     "pykube-ng>=23.6.0",