kubernetes_asyncio.watch package

Submodules

• kubernetes_asyncio.watch.watch module
  • Stream
  • Watch
    • Watch.close()
    • Watch.get_return_type()
    • Watch.get_watch_argument_name()
    • Watch.next()
    • Watch.stop()
    • Watch.stream()
    • Watch.unmarshal_event()
• kubernetes_asyncio.watch.watch_test module
  • WatchTest
    • WatchTest.test_unmarshal_bookmark_succeeds_and_preserves_resource_version()
    • WatchTest.test_unmarshal_job_bookmark_succeeds_and_preserves_resource_version()
    • WatchTest.test_unmarshal_with_custom_object()
    • WatchTest.test_unmarshal_with_empty_return_type()
    • WatchTest.test_unmarshal_with_float_object()
    • WatchTest.test_unmarshal_without_return_type()
    • WatchTest.test_unmarshall_job_bookmark_malformed_object_fails()
    • WatchTest.test_unmarshall_k8s_error_response()
    • WatchTest.test_unmarshall_k8s_error_response_401_gke()
    • WatchTest.test_watch_for_follow()
    • WatchTest.test_watch_k8s_empty_response()
    • WatchTest.test_watch_retry_410()
    • WatchTest.test_watch_retry_timeout()
    • WatchTest.test_watch_timeout_with_resource_version()
    • WatchTest.test_watch_with_decode()
    • WatchTest.test_watch_with_exception()

Module contents

class kubernetes_asyncio.watch.Watch(return_type: str | None = None)

Bases: object

async close() → None

get_return_type(func: object) → str

get_watch_argument_name(func: object) → str

async next() → Any

stop() → None

stream(func, *args, **kwargs) → Watch

Watch an API resource and stream the result back via a generator.

Parameters:

func – The API function pointer. Any parameter to the function can be passed after this parameter.

Returns:

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_asyncio.client.CoreV1Api() watch = kubernetes_asyncio.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()

unmarshal_event(data: str | bytes, response_type: str | None) → Any

Return the K8s response data in JSON format.