diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index a4a2420..e01d4dd 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -35,4 +35,4 @@ repos: rev: v3.19.1 hooks: - id: pyupgrade - args: [--py38-plus] + args: [--py39-plus] diff --git a/examples/kubectl-ng/kubectl_ng/_config.py b/examples/kubectl-ng/kubectl_ng/_config.py index a72cd98..41b9592 100644 --- a/examples/kubectl-ng/kubectl_ng/_config.py +++ b/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 diff --git a/examples/kubectl-ng/kubectl_ng/_exec.py b/examples/kubectl-ng/kubectl_ng/_exec.py index 7fd1074..3d4f849 100644 --- a/examples/kubectl-ng/kubectl_ng/_exec.py +++ b/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) diff --git a/examples/kubectl-ng/kubectl_ng/_get.py b/examples/kubectl-ng/kubectl_ng/_get.py index e231808..b9d13ea 100644 --- a/examples/kubectl-ng/kubectl_ng/_get.py +++ b/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", diff --git a/examples/kubectl-ng/kubectl_ng/_wait.py b/examples/kubectl-ng/kubectl_ng/_wait.py index f4bfec6..6fda5ab 100644 --- a/examples/kubectl-ng/kubectl_ng/_wait.py +++ b/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: diff --git a/kr8s/__init__.py b/kr8s/__init__.py index 3fd6afe..39b1d59 100644 --- a/kr8s/__init__.py +++ b/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 diff --git a/kr8s/_api.py b/kr8s/_api.py index e193a83..362da92 100644 --- a/kr8s/_api.py +++ b/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 diff --git a/kr8s/_async_utils.py b/kr8s/_async_utils.py index ec53224..2165526 100644 --- a/kr8s/_async_utils.py +++ b/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, ) diff --git a/kr8s/_config.py b/kr8s/_config.py index 0df9146..9175b4d 100644 --- a/kr8s/_config.py +++ b/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 [] diff --git a/kr8s/_exec.py b/kr8s/_exec.py index 8637665..9ed9069 100644 --- a/kr8s/_exec.py +++ b/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 diff --git a/kr8s/_objects.py b/kr8s/_objects.py index 473c0d9..4e80b05 100644 --- a/kr8s/_objects.py +++ b/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): diff --git a/kr8s/_portforward.py b/kr8s/_portforward.py index a986a3c..5317850 100644 --- a/kr8s/_portforward.py +++ b/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 diff --git a/kr8s/_testutils.py b/kr8s/_testutils.py index 2caa838..ef4f647 100644 --- a/kr8s/_testutils.py +++ b/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 diff --git a/kr8s/_types.py b/kr8s/_types.py index 471d1eb..2616ba8 100644 --- a/kr8s/_types.py +++ b/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 diff --git a/kr8s/asyncio/_helpers.py b/kr8s/asyncio/_helpers.py index c3dabbe..517c59a 100644 --- a/kr8s/asyncio/_helpers.py +++ b/kr8s/asyncio/_helpers.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: Copyright (c) 2023-2025, Kr8s Developers (See LICENSE for list) # SPDX-License-Identifier: BSD 3-Clause License -from typing import Dict, Optional, Type, Union +from typing import Optional, Union from kr8s._api import Api from kr8s._objects import APIObject @@ -12,9 +12,9 @@ async def get( kind: str, *names: str, namespace: Optional[str] = None, - label_selector: Optional[Union[str, Dict]] = None, - field_selector: Optional[Union[str, Dict]] = None, - as_object: Optional[Type[APIObject]] = None, + label_selector: Optional[Union[str, dict]] = None, + field_selector: Optional[Union[str, dict]] = None, + as_object: Optional[type[APIObject]] = None, allow_unknown_type: bool = True, api=None, _asyncio=True, @@ -77,8 +77,8 @@ async def version(api=None, _asyncio=True): async def watch( kind: str, namespace: Optional[str] = None, - label_selector: Optional[Union[str, Dict]] = None, - field_selector: Optional[Union[str, Dict]] = None, + label_selector: Optional[Union[str, dict]] = None, + field_selector: Optional[Union[str, dict]] = None, since: Optional[str] = None, api=None, _asyncio=True,