From d7ddc2d655133de51f2dd9feb8bc82dbff5dede5 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Wed, 22 Jul 2026 12:33:28 -0400 Subject: [PATCH] [watch] type stream events Type kubernetes.watch events using the watched list method or an explicit model class. Preserve bookmark, raw, log, and argument-forwarding behavior in focused type-checker coverage. Anchor the generated-package ignore rule so the new watch stub sources are included in the built distribution. --- .gitignore | 2 +- .../base/kubernetes-stubs/watch/__init__.pyi | 1 + codegen/base/kubernetes-stubs/watch/watch.pyi | 96 +++++++++++++++++++ tests/typecheck.py | 28 ++++++ 4 files changed, 126 insertions(+), 1 deletion(-) create mode 100644 codegen/base/kubernetes-stubs/watch/__init__.pyi create mode 100644 codegen/base/kubernetes-stubs/watch/watch.pyi create mode 100644 tests/typecheck.py diff --git a/.gitignore b/.gitignore index a799435..dc713bf 100644 --- a/.gitignore +++ b/.gitignore @@ -3,5 +3,5 @@ __pycache__ venv/ kubernetes-client-python/ -kubernetes-stubs/ +/kubernetes-stubs/ kubernetes_ext/ diff --git a/codegen/base/kubernetes-stubs/watch/__init__.pyi b/codegen/base/kubernetes-stubs/watch/__init__.pyi new file mode 100644 index 0000000..a0d67f5 --- /dev/null +++ b/codegen/base/kubernetes-stubs/watch/__init__.pyi @@ -0,0 +1 @@ +from .watch import Watch as Watch diff --git a/codegen/base/kubernetes-stubs/watch/watch.pyi b/codegen/base/kubernetes-stubs/watch/watch.pyi new file mode 100644 index 0000000..33c2be4 --- /dev/null +++ b/codegen/base/kubernetes-stubs/watch/watch.pyi @@ -0,0 +1,96 @@ +import typing + +_ReturnMode = typing.TypeVar("_ReturnMode") +_P = typing.ParamSpec("_P") +_T = typing.TypeVar("_T") +_ListItemT_co = typing.TypeVar("_ListItemT_co", covariant=True) + +# Watch deserializes ordinary events to the inferred model, but leaves bookmark +# objects untouched: +# https://github.com/kubernetes-client/python/blob/01b1642b835f5195a8f81a788908f2cb3991c671/kubernetes/base/watch/watch.py#L119-L149 +class _ObjectEvent(typing.TypedDict, typing.Generic[_T]): + type: typing.Literal["ADDED", "MODIFIED", "DELETED"] + raw_object: object + object: _T + +class _BookmarkEvent(typing.TypedDict): + type: typing.Literal["BOOKMARK"] + raw_object: object + object: object + +_DecodedWatchItem = typing.Union[_ObjectEvent[_T], _BookmarkEvent, None] + +# Without an explicit return type, Watch drops `List` from the API return type; +# generated list stubs expose the same relationship through `items`: +# https://github.com/kubernetes-client/python/blob/01b1642b835f5195a8f81a788908f2cb3991c671/kubernetes/base/watch/watch.py#L105-L111 +class _WatchList(typing.Protocol[_ListItemT_co]): + @property + def items(self) -> typing.Sequence[_ListItemT_co]: ... + +# ApiClient accepts either a class literal or a class-name string as an +# explicit return type: +# https://github.com/kubernetes-client/python/blob/01b1642b835f5195a8f81a788908f2cb3991c671/kubernetes/client/api_client.py#L265-L285 +class _InferReturn: ... +class _FixedReturn(typing.Generic[_T]): ... + +class Watch(typing.Generic[_ReturnMode]): + resource_version: typing.Optional[str] + @typing.overload + def __init__(self: Watch[_InferReturn], return_type: None = ...) -> None: ... + @typing.overload + def __init__(self: Watch[_FixedReturn[_T]], return_type: type[_T]) -> None: ... + # Arbitrary class-name strings cannot be resolved statically. + @typing.overload + def __init__(self: Watch[_FixedReturn[object]], return_type: str) -> None: ... + def stop(self) -> None: ... + # Watch() infers the event object type from the API method's list type. + # deserialize is consumed by Watch rather than forwarded to func. False + # bypasses model deserialization, so an explicit value has a broad result: + # https://github.com/kubernetes-client/python/blob/01b1642b835f5195a8f81a788908f2cb3991c671/kubernetes/base/watch/watch.py#L185-L210 + # PEP 612 cannot express a keyword consumed before forwarding a ParamSpec, + # so the explicit overload intentionally overlaps ordinary calls. + @typing.overload + def stream( # type: ignore[overload-overlap] + self: Watch[_InferReturn], + func: typing.Callable[..., object], + *args: object, + deserialize: bool, + **kwargs: object, + ) -> typing.Iterator[object]: ... + @typing.overload + def stream( + self: Watch[_InferReturn], + func: typing.Callable[_P, _WatchList[_T]], + *args: _P.args, + **kwargs: _P.kwargs, + ) -> typing.Iterator[_DecodedWatchItem[_T]]: ... + @typing.overload + def stream( + self: Watch[_InferReturn], + func: typing.Callable[_P, object], + *args: _P.args, + **kwargs: _P.kwargs, + ) -> typing.Iterator[object]: ... + # Watch(return_type) uses that type instead of the API method's item type. + @typing.overload + def stream( # type: ignore[overload-overlap] + self: Watch[_FixedReturn[_T]], + func: typing.Callable[..., object], + *args: object, + deserialize: bool, + **kwargs: object, + ) -> typing.Iterator[object]: ... + @typing.overload + def stream( + self: Watch[_FixedReturn[_T]], + func: typing.Callable[_P, _WatchList[object]], + *args: _P.args, + **kwargs: _P.kwargs, + ) -> typing.Iterator[_DecodedWatchItem[_T]]: ... + @typing.overload + def stream( + self: Watch[_FixedReturn[_T]], + func: typing.Callable[_P, object], + *args: _P.args, + **kwargs: _P.kwargs, + ) -> typing.Iterator[object]: ... diff --git a/tests/typecheck.py b/tests/typecheck.py new file mode 100644 index 0000000..e4cc647 --- /dev/null +++ b/tests/typecheck.py @@ -0,0 +1,28 @@ +from typing import assert_type + +from kubernetes import client +from kubernetes.watch import Watch + +api = client.CoreV1Api() +inferred_item = next(Watch().stream(api.list_namespaced_pod, "default")) +# ty's generic-protocol constraint solving still produces Unknown here: +# https://github.com/astral-sh/ty/issues/623 +if inferred_item is not None: + if inferred_item["type"] == "BOOKMARK": + bookmark_object = inferred_item["object"] + assert_type(bookmark_object, object) # ty: ignore[type-assertion-failure] + else: + pod_object = inferred_item["object"] + assert_type(pod_object, client.V1Pod) # ty: ignore[type-assertion-failure] + +fixed_item = next(Watch(client.V1Pod).stream(api.list_namespaced_config_map, "default")) +if fixed_item is not None and fixed_item["type"] != "BOOKMARK": + fixed_object = fixed_item["object"] + assert_type(fixed_object, client.V1Pod) # ty: ignore[type-assertion-failure] + +raw_item = next(Watch().stream(api.list_namespaced_pod, "default", deserialize=False)) +assert_type(raw_item, object) +log_item = next(Watch().stream(api.read_namespaced_pod_log, "pod", "default")) +assert_type(log_item, object) + +Watch().stream(api.list_namespaced_pod) # type: ignore[call-overload] # ty: ignore[no-matching-overload]