krkn_lib.k8s.kubernetes_object_helpers module

Kubernetes object helper for unified resource retrieval.

Provides a central place for API mappings and shared logic for fetching Kubernetes objects, with pagination support for large-scale deployments via the continue token mechanism. Supports watching CRD objects for real-time event monitoring.

class krkn_lib.k8s.kubernetes_object_helpers.KubernetesObjectHelpers(krkn)

Bases: object

Helper class for retrieving Kubernetes objects of specific types.

CLUSTER_SCOPED_API_MAP = {'node': ('cli', 'read_node'), 'persistentvolume': ('cli', 'read_persistent_volume')}
CLUSTER_SCOPED_LIST_API_MAP = {'node': ('cli', 'list_node'), 'persistentvolume': ('cli', 'list_persistent_volume')}
NAMESPACED_API_MAP = {'cronjob': ('batch_cli', 'read_namespaced_cron_job'), 'daemonset': ('apps_api', 'read_namespaced_daemon_set'), 'deployment': ('apps_api', 'read_namespaced_deployment'), 'job': ('batch_cli', 'read_namespaced_job'), 'persistentvolumeclaim': ('cli', 'read_namespaced_persistent_volume_claim'), 'pod': ('cli', 'read_namespaced_pod'), 'replicaset': ('apps_api', 'read_namespaced_replica_set'), 'service': ('cli', 'read_namespaced_service'), 'statefulset': ('apps_api', 'read_namespaced_stateful_set')}
NAMESPACED_LIST_API_MAP = {'cronjob': ('batch_cli', 'list_namespaced_cron_job'), 'daemonset': ('apps_api', 'list_namespaced_daemon_set'), 'deployment': ('apps_api', 'list_namespaced_deployment'), 'job': ('batch_cli', 'list_namespaced_job'), 'persistentvolumeclaim': ('cli', 'list_namespaced_persistent_volume_claim'), 'pod': ('cli', 'list_namespaced_pod'), 'replicaset': ('apps_api', 'list_namespaced_replica_set'), 'service': ('cli', 'list_namespaced_service'), 'statefulset': ('apps_api', 'list_namespaced_stateful_set')}
__init__(krkn)

Initialize with a KrknKubernetes instance.

get_object_by_name(kind: str, name: str, namespace: str = None) → Dict | None

Get a Kubernetes object by kind and name, returning it as a dictionary.

This is a universal helper that works with any supported Kubernetes resource type, handling both namespaced and cluster-scoped resources automatically.

Supported resource kinds: - Namespaced: Pod, Deployment, StatefulSet, DaemonSet, ReplicaSet, Service,

PersistentVolumeClaim, Job, CronJob

  • Cluster-scoped: Node, PersistentVolume

Parameters:

  • kind – Kubernetes resource kind (e.g., “Pod”, “Deployment”, “Node”)

  • name – Name of the object

  • namespace – Namespace (required for namespaced resources, ignored for cluster-scoped)

Returns:

Object as a dictionary, or None if not found or unsupported kind

Raises:

ApiException – If the API call fails (e.g., object not found)

list_objects_by_kind(kind: str, namespace: str = None, label_selector: str = None, limit: int = None) → List[Dict]

List all Kubernetes objects of a specific kind with pagination support.

Handles large-scale deployments by using the continue token mechanism to paginate through results when they exceed the limit. Uses krkn’s list_continue_helper for consistent pagination behavior.

Supported resource kinds: - Namespaced: Pod, Deployment, StatefulSet, DaemonSet, ReplicaSet, Service,

PersistentVolumeClaim, Job, CronJob

  • Cluster-scoped: Node, PersistentVolume

Parameters:

  • kind – Kubernetes resource kind (e.g., “Pod”, “Deployment”, “Node”)

  • namespace – Namespace (required for namespaced resources)

  • label_selector – Label selector for filtering (optional)

  • limit – Max items per API request (optional, uses krkn default if not set)

Returns:

List of objects as dictionaries

Raises:

ValueError – If namespace required but not provided

watch_crd_objects(group: str, version: str, plural: str, namespace: str = None, name: str = None, label_selector: str = None, field_selector: str = None, timeout_seconds: int = None, event_handler: Callable[[str, Dict], None] = None) → List[Dict] | None

Watch CRD objects for real-time events (added, modified, deleted).

Can be used to monitor custom resources as they change. Yields events until timeout or explicitly stopped.

Parameters:

  • group – API group of the CRD (e.g., “example.com”)

  • version – API version (e.g., “v1”, “v1alpha1”)

  • plural – Plural name of the custom resource (e.g., “widgets”)

  • namespace – Namespace for namespaced CRDs (optional)

  • name – Watch specific resource by name (optional)

  • label_selector – Watch resources matching labels (optional)

  • field_selector – Watch resources matching fields (optional)

  • timeout_seconds – Watch timeout in seconds (optional)

  • event_handler – Callback function(event_type, object) for each event (optional)

Returns:

List of all objects seen during watch (or None if handler is provided)

Example:
def handle_event(event_type, obj):

print(f”{event_type}: {obj[‘metadata’][‘name’]}”)

helpers.watch_crd_objects(

“example.com”, “v1”, “widgets”, namespace=”default”, event_handler=handle_event, timeout_seconds=60

)