From 53a96320a2f15dc14fb704bd5e1b746134a18122 Mon Sep 17 00:00:00 2001 From: yliao Date: Thu, 9 Jul 2026 23:05:18 +0000 Subject: [PATCH 1/4] update changelog with release notes from master branch --- CHANGELOG.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d662cedb5..51a1c42f7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,23 @@ +# v36.0.3 + +Kubernetes API Version: v1.36.2 + +### Bug or Regression +- Fix Watch.stream selecting watch instead of follow when streaming pod logs. +- Start the leader election worker thread as a daemon so it does not block process shutdown. +- Fix readline_channel, readline_stdout and readline_stderr crashing with OverflowError when using the default timeout, and returning None when a timeout expires. +- Fix format_quantity returning imprecise, non-canonical values for the milli, micro and nano suffixes, and ignoring quantize=Decimal(0). + +# v36.0.2 + +Kubernetes API Version: v1.36.1 + +### Uncategorized +- Restored backward compatibility for `Configuration.auth_settings()`: + the legacy `api_key['authorization']` lookup is honored as a fallback + when `api_key['BearerToken']` is not set, fixing 401 Unauthorized + regressions seen after upgrading to v36.0.0 (#2595). (#2604, @GK-07) + # v36.0.1 Kubernetes API Version: v1.36.1 From 4119fe7b3d21890d9f2aac40ddd3bfe704d01b9f Mon Sep 17 00:00:00 2001 From: yliao Date: Mon, 13 Jul 2026 21:07:35 +0000 Subject: [PATCH 2/4] updated compatibility matrix --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 63a9c306a2..65b1de26ac 100644 --- a/README.md +++ b/README.md @@ -134,7 +134,7 @@ supported versions of Kubernetes clusters. - [client 33.y.z](https://pypi.org/project/kubernetes/33.1.0/): Kubernetes 1.32 or below (+-), Kubernetes 1.33 (✓), Kubernetes 1.34 or above (+-) - [client 34.y.z](https://pypi.org/project/kubernetes/34.1.0/): Kubernetes 1.33 or below (+-), Kubernetes 1.34 (✓), Kubernetes 1.35 or above (+-) - [client 35.y.z](https://pypi.org/project/kubernetes/35.0.0/): Kubernetes 1.34 or below (+-), Kubernetes 1.35 (✓), Kubernetes 1.36 or above (+-) -- [client 36.y.z](https://pypi.org/project/kubernetes/36.0.1/): Kubernetes 1.35 or below (+-), Kubernetes 1.36 (✓), Kubernetes 1.37 or above (+-) +- [client 36.y.z](https://pypi.org/project/kubernetes/36.0.3/): Kubernetes 1.35 or below (+-), Kubernetes 1.36 (✓), Kubernetes 1.37 or above (+-) > See [here](#homogenizing-the-kubernetes-python-client-versions) for an explanation of why there is no v13-v16 release. From bdaf1ac7bb5ee6975b8d49577a20584b09302b08 Mon Sep 17 00:00:00 2001 From: yliao Date: Tue, 14 Jul 2026 22:12:20 +0000 Subject: [PATCH 3/4] ported watch from /github.com/tomplus/kubernetes_asyncio --- kubernetes/aio/watch/__init__.py | 15 + kubernetes/aio/watch/watch.py | 257 +++++++++++++++++ kubernetes/aio/watch/watch_test.py | 425 +++++++++++++++++++++++++++++ 3 files changed, 697 insertions(+) create mode 100644 kubernetes/aio/watch/__init__.py create mode 100644 kubernetes/aio/watch/watch.py create mode 100644 kubernetes/aio/watch/watch_test.py diff --git a/kubernetes/aio/watch/__init__.py b/kubernetes/aio/watch/__init__.py new file mode 100644 index 0000000000..ca9ac06987 --- /dev/null +++ b/kubernetes/aio/watch/__init__.py @@ -0,0 +1,15 @@ +# Copyright 2016 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from .watch import Watch diff --git a/kubernetes/aio/watch/watch.py b/kubernetes/aio/watch/watch.py new file mode 100644 index 0000000000..975f6c0a0f --- /dev/null +++ b/kubernetes/aio/watch/watch.py @@ -0,0 +1,257 @@ +# Copyright 2016 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import asyncio +import json +import pydoc +from functools import partial +from types import SimpleNamespace + +from kubernetes.aio import client + +PYDOC_RETURN_LABEL = ":rtype:" +PYDOC_FOLLOW_PARAM = ":param follow:" + +# Removing this suffix from return type name should give us event's object +# type. e.g., if list_namespaces() returns "NamespaceList" type, +# then list_namespaces(watch=true) returns a stream of events with objects +# of type "Namespace". In case this assumption is not true, user should +# provide return_type to Watch class's __init__. +TYPE_LIST_SUFFIX = "List" + + +def _find_return_type(func): + for line in pydoc.getdoc(func).splitlines(): + if line.startswith(PYDOC_RETURN_LABEL): + return line[len(PYDOC_RETURN_LABEL):].strip() + return "" + + +class Stream(object): + + def __init__(self, func, *args, **kwargs): + pass + + +class Watch(object): + + def __init__(self, return_type=None): + self._raw_return_type = return_type + self._stop = False + self._api_client = client.ApiClient() + self.resource_version = None + self.resp = None + + def stop(self): + self._stop = True + + def get_return_type(self, func): + if self._raw_return_type: + return self._raw_return_type + return_type = _find_return_type(func) + + if return_type.endswith(TYPE_LIST_SUFFIX): + return return_type[:-len(TYPE_LIST_SUFFIX)] + return return_type + + def get_watch_argument_name(self, func): + if PYDOC_FOLLOW_PARAM in pydoc.getdoc(func): + return 'follow' + else: + return 'watch' + + def unmarshal_event(self, data: str, response_type): + """Return the K8s response `data` in JSON format. + + """ + try: + js = json.loads(data) + except ValueError: + return data + + if 'object' not in js or 'type' not in js: + # raise error with code if set + if 'code' in js: + reason = "{}: {}".format(js.get('reason'), js.get('message')) + raise client.exceptions.ApiException(status=js['code'], reason=reason) + + raise Exception(("Malformed JSON response, the 'object' and/or " + "'type' field is missing. JSON: {}").format(js)) + # Make a copy of the original object and save it under the + # `raw_object` key because we will replace the data under `object` with + # a Python native type shortly. + js['raw_object'] = js['object'] + + # Something went wrong. A typical example would be that the user + # supplied a resource version that was too old. In that case K8s would + # not send a conventional ADDED/DELETED/... event but an error. Turn + # this error into a Python exception to save the user the hassle. + if js['type'].lower() == 'error': + obj = js['raw_object'] + reason = "{}: {}".format(obj['reason'], obj['message']) + raise client.exceptions.ApiException(status=obj['code'], reason=reason) + + if js['type'].lower() != 'bookmark': + # If possible, compile the JSON response into a Python native response + # type, eg `V1Namespace` or `V1Pod`,`ExtensionsV1beta1Deployment`, ... + if response_type: + js['object'] = self._api_client.deserialize( + response=SimpleNamespace(data=json.dumps(js['raw_object'])), + response_type=response_type + ) + + # decode and save resource_version to continue watching + if hasattr(js['object'], 'metadata'): + self.resource_version = js['object'].metadata.resource_version + + # For custom objects that we don't have model defined, json + # deserialization results in dictionary + elif (isinstance(js['object'], dict) + and 'metadata' in js['object'] + and 'resourceVersion' in js['object']['metadata']): + self.resource_version = js['object']['metadata']['resourceVersion'] + + elif js['type'].lower() == 'bookmark': + if (isinstance(js['raw_object'], dict) + and 'metadata' in js['raw_object'] + and 'resourceVersion' in js['raw_object']['metadata']): + self.resource_version = js['raw_object']['metadata']['resourceVersion'] + else: + raise Exception(("Malformed JSON response for bookmark event, " + "'metadata' or 'resourceVersion' field is missing. " + "JSON: {}").format(js)) + + return js + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return await self.next() + except: # noqa: E722 + await self.close() + raise + + def _reconnect(self): + self.resp.close() + self.resp = None + if self.resource_version: + self.func.keywords['resource_version'] = self.resource_version + + async def next(self): + + watch_forever = 'timeout_seconds' not in self.func.keywords + retry_410 = watch_forever + + while 1: + + # Set the response object to the user supplied function (eg + # `list_namespaced_pods`) if this is the first iteration. + if self.resp is None: + self.resp = await self.func() + + # Abort at the current iteration if the user has called `stop` on this + # stream instance. + if self._stop: + raise StopAsyncIteration + + # Fetch the next K8s response. + try: + line = await self.resp.content.readline() + except asyncio.TimeoutError: + # This exception can be raised by aiohttp (client timeout) + # but we don't retry if server side timeout is applied. + if watch_forever: + self._reconnect() + continue + else: + raise + + line = line.decode('utf8') + + # Special case for faster log streaming + if self.return_type == 'str': + if line == '': + # end of log + raise StopAsyncIteration + return line + + # Stop the iterator if K8s sends an empty response. This happens when + # eg the supplied timeout has expired. + if line == '': + if watch_forever: + self._reconnect() + continue + raise StopAsyncIteration + + # retry 410 error only once + try: + event = self.unmarshal_event(line, self.return_type) + except client.exceptions.ApiException as ex: + if ex.status == 410 and retry_410: + retry_410 = False # retry only once + self._reconnect() + continue + raise + retry_410 = watch_forever + return event + + def stream(self, func, *args, **kwargs): + """Watch an API resource and stream the result back via a generator. + + :param func: The API function pointer. Any parameter to the function + can be passed after this parameter. + + :return: Event object with these keys: + 'type': The type of event such as "ADDED", "DELETED", etc. + 'raw_object': a dict representing the watched object. + 'object': A model representation of raw_object. The name of + model will be determined based on + the func's doc string. If it cannot be determined, + 'object' value will be the same as 'raw_object'. + + Example: + v1 = kubernetes.aio.client.CoreV1Api() + watch = kubernetes.aio.watch.Watch() + async for e in watch.stream(v1.list_namespace, timeout_seconds=10): + type = e['type'] + object = e['object'] # object is one of type return_type + raw_object = e['raw_object'] # raw_object is a dict + ... + if should_stop: + watch.stop() + """ + self._stop = False + self.return_type = self.get_return_type(func) + kwargs[self.get_watch_argument_name(func)] = True + kwargs['_preload_content'] = False + if 'resource_version' in kwargs: + self.resource_version = kwargs['resource_version'] + + self.func = partial(func, *args, **kwargs) + + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_value, traceback): + await self.close() + + async def close(self): + await self._api_client.close() + if self.resp is not None: + self.resp.release() + self.resp = None diff --git a/kubernetes/aio/watch/watch_test.py b/kubernetes/aio/watch/watch_test.py new file mode 100644 index 0000000000..a73187c69a --- /dev/null +++ b/kubernetes/aio/watch/watch_test.py @@ -0,0 +1,425 @@ +# Copyright 2016 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import asyncio +import json +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, Mock, call + +import kubernetes.aio +from kubernetes.aio.watch import Watch + + +class WatchTest(IsolatedAsyncioTestCase): + + async def test_watch_with_decode(self): + fake_resp = AsyncMock() + fake_resp.content.readline = AsyncMock() + fake_resp.release = Mock() + side_effects = [ + { + "type": "ADDED", + "object": { + "metadata": {"name": "test{}".format(uid), + "resourceVersion": str(uid)}, + "spec": {}, "status": {} + } + } + for uid in range(3) + ] + side_effects = [json.dumps(_).encode('utf8') for _ in side_effects] + side_effects.extend([AssertionError('Should not have been called')]) + fake_resp.content.readline.side_effect = side_effects + + fake_api = Mock() + fake_api.get_namespaces = AsyncMock(return_value=fake_resp) + fake_api.get_namespaces.__doc__ = ':rtype: V1NamespaceList' + + watch = kubernetes.aio.watch.Watch() + count = 0 + async with watch: + async for e in watch.stream(fake_api.get_namespaces, resource_version='123'): + self.assertEqual("ADDED", e['type']) + # make sure decoder worked and we got a model with the right name + self.assertEqual("test%d" % count, e['object'].metadata.name) + # make sure decoder worked and updated Watch.resource_version + self.assertEqual(e['object'].metadata.resource_version, str(count)) + self.assertEqual(watch.resource_version, str(count)) + + # Stop the watch. This must not return the next event which would + # be an AssertionError exception. + count += 1 + if count == len(side_effects) - 1: + watch.stop() + + fake_api.get_namespaces.assert_called_once_with( + _preload_content=False, watch=True, resource_version='123') + fake_resp.release.assert_called_once_with() + + # last resource_version has to be stored in the object + self.assertEqual(watch.resource_version, '2') + + async def test_watch_for_follow(self): + fake_resp = AsyncMock() + fake_resp.content.readline = AsyncMock() + fake_resp.release = Mock() + side_effects = ['log_line_1', 'log_line_2', ''] + side_effects = [_.encode('utf8') for _ in side_effects] + side_effects.extend([AssertionError('Should not have been called')]) + fake_resp.content.readline.side_effect = side_effects + + fake_api = Mock() + fake_api.read_namespaced_pod_log = AsyncMock(return_value=fake_resp) + fake_api.read_namespaced_pod_log.__doc__ = ':param follow:\n:type follow: bool\n:rtype: str' + + watch = kubernetes.aio.watch.Watch() + logs = [] + async with watch: + async for e in watch.stream(fake_api.read_namespaced_pod_log): + logs.append(e) + + self.assertListEqual(logs, ['log_line_1', 'log_line_2']) + fake_api.read_namespaced_pod_log.assert_called_once_with( + _preload_content=False, follow=True) + fake_resp.release.assert_called_once_with() + + async def test_watch_k8s_empty_response(self): + """Stop the iterator when the response is empty. + + This typically happens when the user supplied timeout expires. + + """ + # Mock the readline return value to first return a valid response + # followed by an empty response. + fake_resp = AsyncMock() + fake_resp.content.readline = AsyncMock() + side_effects = [ + {"type": "ADDED", "object": {"metadata": {"name": "test0"}, "spec": {}, "status": {}}}, + {"type": "ADDED", "object": {"metadata": {"name": "test1"}, "spec": {}, "status": {}}}, + ] + side_effects = [json.dumps(_).encode('utf8') for _ in side_effects] + fake_resp.content.readline.side_effect = side_effects + [b''] + + # Fake the K8s resource object to watch. + fake_api = Mock() + fake_api.get_namespaces = AsyncMock(return_value=fake_resp) + fake_api.get_namespaces.__doc__ = ':rtype: V1NamespaceList' + + # Iteration must cease after all valid responses were received. + watch = kubernetes.aio.watch.Watch() + cnt = 0 + async for _ in watch.stream(fake_api.get_namespaces): # noqa + cnt += 1 + self.assertEqual(cnt, len(side_effects)) + + async def test_unmarshal_with_float_object(self): + w = Watch() + event = w.unmarshal_event('{"type": "ADDED", "object": 1}', 'float') + self.assertEqual("ADDED", event['type']) + self.assertEqual(1.0, event['object']) + self.assertTrue(isinstance(event['object'], float)) + self.assertEqual(1, event['raw_object']) + + async def test_unmarshal_without_return_type(self): + w = Watch() + event = w.unmarshal_event( + '{"type": "ADDED", "object": ["test1"]}', None) + self.assertEqual("ADDED", event['type']) + self.assertEqual(["test1"], event['object']) + self.assertEqual(["test1"], event['raw_object']) + + async def test_unmarshal_with_empty_return_type(self): + # empty string as a return_type is a default value + # if watch can't detect object by function's name + w = Watch() + event = w.unmarshal_event( + '{"type": "ADDED", "object": ["test1"]}', '') + self.assertEqual("ADDED", event['type']) + self.assertEqual(["test1"], event['object']) + self.assertEqual(["test1"], event['raw_object']) + + async def test_unmarshall_k8s_error_response(self): + """Never parse messages of type ERROR. + + This test uses an actually recorded error, in this case for an outdated + resource version. + + """ + # An actual error response sent by K8s during testing. + k8s_err = { + 'type': 'ERROR', + 'object': { + 'kind': 'Status', + 'apiVersion': 'v1', + 'metadata': {}, + 'status': 'Failure', + 'message': 'too old resource version: 1 (8146471)', + 'reason': 'Gone', + 'code': 410 + } + } + + with self.assertRaisesRegex( + kubernetes.aio.client.exceptions.ApiException, + r'\(410\)\nReason: Gone: too old resource version: 1 \(8146471\)'): + Watch().unmarshal_event(json.dumps(k8s_err), None) + + async def test_unmarshall_k8s_error_response_401_gke(self): + """Never parse messages of type ERROR. + + This test uses an actually recorded error returned by GKE. + + """ + # An actual error response sent by K8s during testing. + k8s_err = { + 'kind': 'Status', + 'apiVersion': 'v1', + 'metadata': {}, + 'status': 'Failure', + 'message': 'Unauthorized', + 'reason': 'Unauthorized', + 'code': 401 + } + + with self.assertRaisesRegex( + kubernetes.aio.client.exceptions.ApiException, + r'\(401\)\nReason: Unauthorized: Unauthorized'): + Watch().unmarshal_event(json.dumps(k8s_err), None) + + async def test_unmarshal_with_custom_object(self): + w = Watch() + event = w.unmarshal_event('{"type": "ADDED", "object": {"apiVersion":' + '"test.com/v1beta1","kind":"foo","metadata":' + '{"name": "bar", "resourceVersion": "1"}}}', + 'object') + self.assertEqual("ADDED", event['type']) + # make sure decoder deserialized json into dictionary and updated + # Watch.resource_version + self.assertTrue(isinstance(event['object'], dict)) + self.assertEqual("1", event['object']['metadata']['resourceVersion']) + self.assertEqual("1", w.resource_version) + + async def test_watch_with_exception(self): + fake_resp = AsyncMock() + fake_resp.content.readline = AsyncMock() + fake_resp.content.readline.side_effect = KeyError("expected") + fake_api = Mock() + fake_api.get_namespaces = AsyncMock(return_value=fake_resp) + fake_api.get_namespaces.__doc__ = ':rtype: V1NamespaceList' + + with self.assertRaises(KeyError): + watch = kubernetes.aio.watch.Watch() + async for e in watch.stream(fake_api.get_namespaces, timeout_seconds=10): # noqa + pass + + async def test_watch_retry_timeout(self): + fake_resp = AsyncMock() + fake_resp.content.readline = AsyncMock() + fake_resp.release = Mock() + + mock_event = {"type": "ADDED", + "object": {"metadata": {"name": "test1555", + "resourceVersion": "1555"}, + "spec": {}, + "status": {}}} + + fake_resp.content.readline.side_effect = [json.dumps(mock_event).encode('utf8'), + asyncio.TimeoutError(), + b""] + + fake_api = Mock() + fake_api.get_namespaces = AsyncMock(return_value=fake_resp) + fake_api.get_namespaces.__doc__ = ':rtype: V1NamespaceList' + + watch = kubernetes.aio.watch.Watch() + async with watch.stream(fake_api.get_namespaces) as stream: + async for e in stream: # noqa + pass + + fake_api.get_namespaces.assert_has_calls( + [call(_preload_content=False, watch=True), + call(_preload_content=False, watch=True, resource_version='1555')]) + fake_resp.release.assert_called_once_with() + + async def test_watch_retry_410(self): + fake_resp = AsyncMock() + fake_resp.content.readline = AsyncMock() + fake_resp.release = Mock() + + mock_event1 = { + "type": "ADDED", + "object": { + "metadata": + { + "name": "test1555", + "resourceVersion": "1555" + }, + "spec": {}, + "status": {} + } + } + + mock_event2 = { + "type": "ADDED", + "object": { + "metadata": + { + "name": "test1555", + "resourceVersion": "1555" + }, + "spec": {}, + "status": {} + } + } + + mock_410 = { + 'type': 'ERROR', + 'object': { + 'kind': 'Status', + 'apiVersion': 'v1', + 'metadata': {}, + 'status': 'Failure', + 'message': 'too old resource version: 1 (8146471)', + 'reason': 'Gone', + 'code': 410 + } + } + + # retry 410 + fake_resp.content.readline.side_effect = [json.dumps(mock_event1).encode('utf8'), + json.dumps(mock_410).encode('utf8'), + json.dumps(mock_event2).encode('utf8'), + json.dumps(mock_410).encode('utf8'), + b""] + + fake_api = Mock() + fake_api.get_namespaces = AsyncMock(return_value=fake_resp) + fake_api.get_namespaces.__doc__ = ':rtype: V1NamespaceList' + + watch = kubernetes.aio.watch.Watch() + async with watch.stream(fake_api.get_namespaces) as stream: + async for e in stream: # noqa + pass + + fake_api.get_namespaces.assert_has_calls( + [call(_preload_content=False, watch=True), + call(_preload_content=False, watch=True, resource_version='1555')]) + fake_resp.release.assert_called_once_with() + + # retry 410 only once + fake_resp.content.readline.side_effect = [json.dumps(mock_event1).encode('utf8'), + json.dumps(mock_410).encode('utf8'), + json.dumps(mock_event2).encode('utf8'), + json.dumps(mock_410).encode('utf8'), + json.dumps(mock_410).encode('utf8'), + b""] + + fake_api = Mock() + fake_api.get_namespaces = AsyncMock(return_value=fake_resp) + fake_api.get_namespaces.__doc__ = ':rtype: V1NamespaceList' + + with self.assertRaisesRegex( + kubernetes.aio.client.exceptions.ApiException, + r'\(410\)\nReason: Gone: too old resource version: 1 \(8146471\)'): + watch = kubernetes.aio.watch.Watch() + async with watch.stream(fake_api.get_namespaces) as stream: + async for e in stream: # noqa + pass + + # no retry 410 if timeout is passed + fake_resp.content.readline.side_effect = [json.dumps(mock_event1).encode('utf8'), + json.dumps(mock_410).encode('utf8'), + b""] + + fake_api = Mock() + fake_api.get_namespaces = AsyncMock(return_value=fake_resp) + fake_api.get_namespaces.__doc__ = ':rtype: V1NamespaceList' + + with self.assertRaisesRegex( + kubernetes.aio.client.exceptions.ApiException, + r'\(410\)\nReason: Gone: too old resource version: 1 \(8146471\)'): + watch = kubernetes.aio.watch.Watch() + async with watch.stream(fake_api.get_namespaces, timeout_seconds=10) as stream: + async for e in stream: # noqa + pass + + async def test_watch_timeout_with_resource_version(self): + fake_resp = AsyncMock() + fake_resp.content.readline = AsyncMock() + fake_resp.release = Mock() + + fake_resp.content.readline.side_effect = [asyncio.TimeoutError(), + b""] + + fake_api = Mock() + fake_api.get_namespaces = AsyncMock(return_value=fake_resp) + fake_api.get_namespaces.__doc__ = ':rtype: V1NamespaceList' + + watch = kubernetes.aio.watch.Watch() + async with watch.stream(fake_api.get_namespaces, resource_version='10') as stream: + async for e in stream: # noqa + pass + + # all calls use the passed resource version + fake_api.get_namespaces.assert_has_calls( + [call(_preload_content=False, watch=True, resource_version='10'), + call(_preload_content=False, watch=True, resource_version='10')]) + + fake_resp.release.assert_called_once_with() + self.assertEqual(watch.resource_version, '10') + + async def test_unmarshal_bookmark_succeeds_and_preserves_resource_version(self): + w = Watch() + event = w.unmarshal_event('{"type": "BOOKMARK", "object": {"apiVersion":' + '"test.com/v1beta1","kind":"foo","metadata":' + '{"name": "bar", "resourceVersion": "1"}}}', + 'object') + self.assertEqual("BOOKMARK", event['type']) + + # make sure the resource version is preserved, + # and the watcher's resource_version is updated + self.assertTrue(isinstance(event['raw_object'], dict)) + self.assertEqual("1", event['raw_object']['metadata']['resourceVersion']) + self.assertEqual("1", w.resource_version) + + async def test_unmarshal_job_bookmark_succeeds_and_preserves_resource_version(self): + w = Watch() + event = w.unmarshal_event('{"type": "BOOKMARK", "object": {"apiVersion":' + '"batch/v1","kind":"Job","metadata":' + '{"name": "bar", "resourceVersion": "1"},' + '"spec": {"template": {"metadata": ' + '{"creationTimestamp":null}, "spec": ' + '{"containers":null}}}}}', + 'object') + self.assertEqual("BOOKMARK", event['type']) + + # make sure the resource version is preserved, + # and the watcher's resource_version is updated + self.assertTrue(isinstance(event['raw_object'], dict)) + self.assertEqual("1", event['raw_object']['metadata']['resourceVersion']) + self.assertEqual("1", w.resource_version) + + async def test_unmarshall_job_bookmark_malformed_object_fails(self): + # An actual error response sent by K8s during testing. + k8s_err = { + 'type': 'BOOKMARK', + 'object': { + 'kind': 'Job', + 'apiVersion': 'batch/v1', + 'metadata': {}, + } + } + + with self.assertRaises(Exception): + Watch().unmarshal_event(json.dumps(k8s_err), None) From eb0367e1d5e9947b7a6f0ead8234b15d49507a22 Mon Sep 17 00:00:00 2001 From: yliao Date: Tue, 14 Jul 2026 22:19:11 +0000 Subject: [PATCH 4/4] ported watch examples from /github.com/tomplus/kubernetes_asyncio --- examples_asyncio/watch_namespaces.py | 31 ++++++++++++++++++ examples_asyncio/watch_ns_pods.py | 49 ++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+) create mode 100644 examples_asyncio/watch_namespaces.py create mode 100644 examples_asyncio/watch_ns_pods.py diff --git a/examples_asyncio/watch_namespaces.py b/examples_asyncio/watch_namespaces.py new file mode 100644 index 0000000000..e67f60e4c7 --- /dev/null +++ b/examples_asyncio/watch_namespaces.py @@ -0,0 +1,31 @@ +import asyncio + +from kubernetes.aio import client, config, watch + + +async def main(): + # Configs can be set in Configuration class directly or using helper + # utility. If no argument provided, the config will be loaded from + # default location. + await config.load_kube_config() + + v1 = client.CoreV1Api() + count = 10 + w = watch.Watch() + + async for event in w.stream(v1.list_namespace, timeout_seconds=10): + print("Event: {} {}".format(event["type"], event["object"].metadata.name)) + count -= 1 + if not count: + w.stop() + + print("Ended.") + # An explicit close is necessary to stop the stream + # or use async context manager like in example4.py + await w.close() + + +if __name__ == "__main__": + loop = asyncio.get_event_loop() + loop.run_until_complete(main()) + loop.close() diff --git a/examples_asyncio/watch_ns_pods.py b/examples_asyncio/watch_ns_pods.py new file mode 100644 index 0000000000..738e65ae63 --- /dev/null +++ b/examples_asyncio/watch_ns_pods.py @@ -0,0 +1,49 @@ +"""Watch multiple K8s event streams without threads.""" + +import asyncio + +from kubernetes.aio import client, config, watch + + +async def watch_namespaces(): + async with client.ApiClient() as api: + v1 = client.CoreV1Api(api) + async with watch.Watch().stream(v1.list_namespace) as stream: + async for event in stream: + etype, obj = event["type"], event["object"] + print("{} namespace {}".format(etype, obj.metadata.name)) + + +async def watch_pods(): + async with client.ApiClient() as api: + v1 = client.CoreV1Api(api) + async with watch.Watch().stream(v1.list_pod_for_all_namespaces) as stream: + async for event in stream: + evt, obj = event["type"], event["object"] + print( + "{} pod {} in NS {}".format( + evt, obj.metadata.name, obj.metadata.namespace + ) + ) + + +def main(): + loop = asyncio.get_event_loop() + + # Load the kubeconfig file specified in the KUBECONFIG environment + # variable, or fall back to `~/.kube/config`. + loop.run_until_complete(config.load_kube_config()) + + # Define the tasks to watch namespaces and pods. + tasks = [ + asyncio.ensure_future(watch_namespaces()), + asyncio.ensure_future(watch_pods()), + ] + + # Push tasks into event loop. + loop.run_until_complete(asyncio.wait(tasks)) + loop.close() + + +if __name__ == "__main__": + main()