Bump pyupgrade minimum to 3.9 (#558)

.pre-commit-config.yaml

@@ -35,4 +35,4 @@ repos:
     rev: v3.19.1
     hooks:
     -   id: pyupgrade
-        args: [--py38-plus]
+        args: [--py39-plus]

examples/kubectl-ng/kubectl_ng/_config.py

@@ -1,12 +1,11 @@
 # SPDX-FileCopyrightText: Copyright (c) 2024-2025, Kr8s Developers (See LICENSE for list)
 # SPDX-License-Identifier: BSD 3-Clause License
-from typing import Optional
+from typing import Annotated, Optional
 
 import rich.table
 import typer
 from rich import box
 from rich.console import Console
-from typing_extensions import Annotated
 
 import kr8s
 

examples/kubectl-ng/kubectl_ng/_exec.py

@@ -1,7 +1,6 @@
 # SPDX-FileCopyrightText: Copyright (c) 2023-2025, Kr8s Developers (See LICENSE for list)
 # SPDX-License-Identifier: BSD 3-Clause License
 import sys
-from typing import List
 
 import typer
 from rich.console import Console
@@ -27,7 +26,7 @@ async def kexec(
         "for selecting the container to be attached or the first container in the "
         "pod will be chosen",
     ),
-    command: List[str] = typer.Argument(..., help="COMMAND [args...]"),
+    command: list[str] = typer.Argument(..., help="COMMAND [args...]"),
 ):
     """Execute a command in a container."""
     pod = await Pod.get(resource, namespace=namespace)

examples/kubectl-ng/kubectl_ng/_get.py

@@ -1,6 +1,5 @@
 # SPDX-FileCopyrightText: Copyright (c) 2023-2025, Kr8s Developers (See LICENSE for list)
 # SPDX-License-Identifier: BSD 3-Clause License
-from typing import List
 
 import anyio
 import rich.table
@@ -62,7 +61,7 @@ async def get_resources(resources, label_selector, field_selector):
 
 
 async def get(
-    resources: List[str] = typer.Argument(..., help="TYPE[.VERSION][.GROUP]"),
+    resources: list[str] = typer.Argument(..., help="TYPE[.VERSION][.GROUP]"),
     all_namespaces: bool = typer.Option(
         False,
         "-A",
@@ -101,7 +100,7 @@ async def get(
         help="When printing, show all labels as the last column "
         "(default hide labels column).",
     ),
-    label_columns: List[str] = typer.Option(
+    label_columns: list[str] = typer.Option(
         [],
         "-L",
         "--label-columns",

examples/kubectl-ng/kubectl_ng/_wait.py

@@ -1,7 +1,7 @@
 # SPDX-FileCopyrightText: Copyright (c) 2023-2025, Kr8s Developers (See LICENSE for list)
 # SPDX-License-Identifier: BSD 3-Clause License
 import asyncio
-from typing import List, Optional
+from typing import Optional
 
 import typer
 from rich.console import Console
@@ -22,7 +22,7 @@
 
 
 async def wait(
-    resources: List[str] = typer.Argument(..., help="TYPE[.VERSION][.GROUP]"),
+    resources: list[str] = typer.Argument(..., help="TYPE[.VERSION][.GROUP]"),
     all_namespaces: bool = typer.Option(
         False,
         "-A",
@@ -40,7 +40,7 @@ async def wait(
         "--all",
         help="Select all resources in the namespace of the specified resource types",
     ),
-    conditions: List[str] = typer.Option(
+    conditions: list[str] = typer.Option(
         [],
         "-f",
         "--for",
@@ -138,7 +138,7 @@ async def wait(
             field_selector=field_selector,
         )
 
-    async def wait_for(o: APIObject, conditions: List[str]):
+    async def wait_for(o: APIObject, conditions: list[str]):
         try:
             await o.wait(conditions=conditions, timeout=timeout)
         except asyncio.TimeoutError:

kr8s/__init__.py

@@ -10,8 +10,8 @@
 # ruff: noqa: D102
 from __future__ import annotations
 
+from collections.abc import Generator
 from functools import partial, update_wrapper
-from typing import Generator
 
 from . import asyncio, objects, portforward
 from ._api import ALL

kr8s/_api.py

@@ -11,9 +11,9 @@
 import threading
 import warnings
 import weakref
+from collections.abc import AsyncGenerator
 from typing import (
     TYPE_CHECKING,
-    AsyncGenerator,
 )
 
 import httpx

kr8s/_async_utils.py

@@ -25,15 +25,13 @@
 import subprocess
 import sys
 import tempfile
+from collections.abc import AsyncGenerator, Awaitable, Generator
 from contextlib import asynccontextmanager
 from functools import partial, wraps
 from threading import Thread
 from typing import (
     Any,
-    AsyncGenerator,
-    Awaitable,
     Callable,
-    Generator,
     TypeVar,
 )
 

kr8s/_config.py

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: BSD 3-Clause License
 import pathlib
 import typing
-from typing import Any, Dict, List, Optional, Protocol, Union
+from typing import Any, Optional, Protocol, Union
 
 import anyio
 import jsonpath
@@ -41,7 +41,7 @@ def get(
 
 
 class KubeConfigSet(KubeConfigMixin):
-    def __init__(self, *paths_or_dicts: Union[PathType, Dict]):
+    def __init__(self, *paths_or_dicts: Union[PathType, dict]):
         self._configs = []
         for path_or_dict in paths_or_dicts:
             try:
@@ -82,7 +82,7 @@ def get_path(self, context: Optional[str] = None) -> PathType:
         return self._configs[0].path
 
     @property
-    def raw(self) -> Dict:
+    def raw(self) -> dict:
         """Merge all kubeconfig data into a single kubeconfig."""
         data = {
             "apiVersion": "v1",
@@ -133,21 +133,21 @@ async def rename_context(self, old: str, new: str) -> None:
                 return
         raise ValueError(f"Context {old} not found")
 
-    def get_context(self, context_name: str) -> Dict:
+    def get_context(self, context_name: str) -> dict:
         """Get a context by name."""
         for context in self.contexts:
             if context["name"] == context_name:
                 return context["context"]
         raise ValueError(f"Context {context_name} not found")
 
-    def get_cluster(self, cluster_name: str) -> Dict:
+    def get_cluster(self, cluster_name: str) -> dict:
         """Get a cluster by name."""
         for cluster in self.clusters:
             if cluster["name"] == cluster_name:
                 return cluster["cluster"]
         raise ValueError(f"Cluster {cluster_name} not found")
 
-    def get_user(self, user_name: str) -> Dict:
+    def get_user(self, user_name: str) -> dict:
         """Get a user by name."""
         for user in self.users:
             if user["name"] == user_name:
@@ -174,11 +174,11 @@ async def unset(self, pointer: str) -> None:
                 pass
 
     @property
-    def preferences(self) -> List[Dict]:
+    def preferences(self) -> list[dict]:
         return self._configs[0].preferences
 
     @property
-    def clusters(self) -> List[Dict]:
+    def clusters(self) -> list[dict]:
         clusters = []
         for config in self._configs:
             if config.clusters:
@@ -189,7 +189,7 @@ def clusters(self) -> List[Dict]:
         return repacked
 
     @property
-    def users(self) -> List[Dict]:
+    def users(self) -> list[dict]:
         users = []
         for config in self._configs:
             if config.users:
@@ -200,7 +200,7 @@ def users(self) -> List[Dict]:
         return repacked
 
     @property
-    def contexts(self) -> List[Dict]:
+    def contexts(self) -> list[dict]:
         contexts = []
         for config in self._configs:
             if config.contexts:
@@ -211,7 +211,7 @@ def contexts(self) -> List[Dict]:
         return repacked
 
     @property
-    def extensions(self) -> List[Dict]:
+    def extensions(self) -> list[dict]:
         extensions = []
         for config in self._configs:
             if config.extensions:
@@ -220,7 +220,7 @@ def extensions(self) -> List[Dict]:
 
 
 class KubeConfig(KubeConfigMixin):
-    def __init__(self, path_or_config: Union[PathType, Dict]):
+    def __init__(self, path_or_config: Union[PathType, dict]):
         self.path: PathType
         self._raw: dict = {}
 
@@ -290,21 +290,21 @@ async def rename_context(self, old: str, new: str) -> None:
                 return
         raise ValueError(f"Context {old} not found")
 
-    def get_context(self, context_name: str) -> Dict:
+    def get_context(self, context_name: str) -> dict:
         """Get a context by name."""
         for context in self.contexts:
             if context["name"] == context_name:
                 return context["context"]
         raise ValueError(f"Context {context_name} not found")
 
-    def get_cluster(self, cluster_name: str) -> Dict:
+    def get_cluster(self, cluster_name: str) -> dict:
         """Get a cluster by name."""
         for cluster in self.clusters:
             if cluster["name"] == cluster_name:
                 return cluster["cluster"]
         raise ValueError(f"Cluster {cluster_name} not found")
 
-    def get_user(self, user_name: str) -> Dict:
+    def get_user(self, user_name: str) -> dict:
         """Get a user by name."""
         for user in self.users:
             if user["name"] == user_name:
@@ -329,25 +329,25 @@ async def unset(self, pointer: str) -> Any:
         await self.save()
 
     @property
-    def raw(self) -> Dict:
+    def raw(self) -> dict:
         return self._raw
 
     @property
-    def preferences(self) -> List[Dict]:
+    def preferences(self) -> list[dict]:
         return self._raw["preferences"]
 
     @property
-    def clusters(self) -> List[Dict]:
+    def clusters(self) -> list[dict]:
         return self._raw["clusters"]
 
     @property
-    def users(self) -> List[Dict]:
+    def users(self) -> list[dict]:
         return self._raw["users"]
 
     @property
-    def contexts(self) -> List[Dict]:
+    def contexts(self) -> list[dict]:
         return self._raw["contexts"]
 
     @property
-    def extensions(self) -> List[Dict]:
+    def extensions(self) -> list[dict]:
         return self._raw["extensions"] if "extensions" in self._raw else []

kr8s/_exec.py

@@ -3,9 +3,10 @@
 from __future__ import annotations
 
 import json
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, AsyncGenerator, BinaryIO
+from typing import TYPE_CHECKING, BinaryIO
 
 from kr8s._exceptions import ExecError
 

kr8s/_objects.py

@@ -7,11 +7,10 @@
 import pathlib
 import re
 import time
+from collections.abc import AsyncGenerator
 from typing import (
     Any,
-    AsyncGenerator,
     BinaryIO,
-    List,
     Literal,
     cast,
 )
@@ -1599,10 +1598,10 @@ async def async_ready_pods(self) -> list[Pod]:
         ]
         if isinstance(pods, Pod):
             pods = [pods]
-        elif isinstance(pods, List) and all(isinstance(pod, Pod) for pod in pods):
+        elif isinstance(pods, list) and all(isinstance(pod, Pod) for pod in pods):
             # The all(isinstance(...) for ...) check doesn't seem to narrow the type
             # correctly in pyright so we need to explicitly use cast
-            pods = cast(List[Pod], pods)
+            pods = cast(list[Pod], pods)
         else:
             raise TypeError(f"Unexpected type {type(pods)} returned from API")
         return [pod for pod in pods if await pod.async_ready()]
@@ -1723,10 +1722,10 @@ async def async_pods(self) -> list[Pod]:
         ]
         if isinstance(pods, Pod):
             return [pods]
-        if isinstance(pods, List) and all(isinstance(pod, Pod) for pod in pods):
+        if isinstance(pods, list) and all(isinstance(pod, Pod) for pod in pods):
             # The all(isinstance(...) for ...) check doesn't seem to narrow the type
             # correctly in pyright so we need to explicitly use cast
-            return cast(List[Pod], pods)
+            return cast(list[Pod], pods)
         raise TypeError(f"Unexpected type {type(pods)} returned from API")
 
     async def ready(self):

kr8s/_portforward.py

@@ -7,8 +7,9 @@
 import random
 import socket
 import sys
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager, suppress
-from typing import TYPE_CHECKING, AsyncGenerator, Literal, Union
+from typing import TYPE_CHECKING, Literal, Union
 
 import anyio
 import httpx_ws

kr8s/_testutils.py

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: BSD 3-Clause License
 import contextlib
 import os
-from typing import Generator
+from collections.abc import Generator
 
 
 @contextlib.contextmanager

kr8s/_types.py

@@ -1,10 +1,9 @@
 # SPDX-FileCopyrightText: Copyright (c) 2024-2025, Kr8s Developers (See LICENSE for list)
 # SPDX-License-Identifier: BSD 3-Clause License
+from collections.abc import Iterable
 from os import PathLike
 from typing import (
     TYPE_CHECKING,
-    Iterable,
-    List,
     Protocol,
     TypeVar,
     Union,
@@ -26,7 +25,7 @@
 class APIObjectWithPods(Protocol):
     """An APIObject subclass that contains other Pod objects."""
 
-    async def async_ready_pods(self) -> List["Pod"]: ...
+    async def async_ready_pods(self) -> list["Pod"]: ...
 
 
 @runtime_checkable