powerk8s

  1"""Powerk8s - a Powerline plugin for Kubernetes information.
  2
  3Powerk8s reads the local $KUBECNOFIG and displays specified information, such
  4as:
  5- Current context
  6- Namespace
  7"""
  8from __future__ import annotations
  9
 10from dataclasses import asdict, dataclass, field
 11from enum import StrEnum
 12from typing import Any, Mapping, Sequence
 13
 14from kubernetes import config  # type: ignore
 15from powerline import PowerlineLogger  # type: ignore
 16
 17KUBERNETES_LOGO: str = "\U00002388 "
 18
 19
 20class SegmentArg(StrEnum):
 21    """All possible Powerline segment argument types for Powerk8s."""
 22
 23    SHOW_KUBERNETES_LOGO = "show_kube_logo"
 24    SHOW_CLUSTER = "show_cluster"
 25    SHOW_NAMESPACE = "show_namespace"
 26    SHOW_DEFAULT_NAMESPACE = "show_default_namespace"
 27
 28
 29class HighlightGroup(StrEnum):
 30    """All possible highlight groups for Powerk8s."""
 31
 32    KUBERNETES_CLUSTER_ALERT = "kubernetes_cluster:alert"
 33    KUBERNETES_CLUSTER = "kubernetes_cluster"
 34    KUBERNETES_DIVIDER = "kubernetes:divider"
 35    KUBERNETES_NAMESPACE_ALERT = "kubernetes_namespace:alert"
 36    KUBERNETES_NAMESPACE = "kubernetes_namespace"
 37
 38
 39@dataclass
 40class SegmentData:
 41    """Encapsulates data for a Powerk8s segment."""
 42
 43    # pylint: disable=unsubscriptable-object
 44    contents: str | None
 45    highlight_groups: Sequence[str]
 46    # pylint: disable=unsubscriptable-object
 47    divider_highlight_group: str | None = field(default="")
 48
 49
 50def get_kubernetes_logo(color: str) -> SegmentData:
 51    """Get the Kubernetes logo (it is hardcoded right now)."""
 52    return SegmentData(
 53        KUBERNETES_LOGO, [color], HighlightGroup.KUBERNETES_DIVIDER.value
 54    )
 55
 56
 57def get_segment_args(**kwargs: Any) -> Mapping[SegmentArg, Any]:
 58    """Get the arguments for a Powerk8s segment."""
 59    return {sa: kwargs.get(sa.value, None) for sa in SegmentArg}
 60
 61
 62def powerk8s(*_: Sequence[Any], **kwargs: Any) -> Sequence[Mapping[str, str]]:
 63    """Entry point to the plugin."""
 64    segment_args: Mapping[SegmentArg, Any] = get_segment_args(**kwargs)
 65
 66    powerline_logger: PowerlineLogger = kwargs.get("pl", None)
 67
 68    _, active_context = config.list_kube_config_contexts()
 69
 70    if powerline_logger is not None:
 71        powerline_logger.debug(f"Context: {active_context}")
 72        powerline_logger.debug(f"Segment arguments: {segment_args}")
 73
 74    segments: list[SegmentData] = []
 75
 76    if segment_args.get(SegmentArg.SHOW_KUBERNETES_LOGO, False):
 77        segments.append(get_kubernetes_logo(HighlightGroup.KUBERNETES_CLUSTER.value))
 78
 79    if segment_args.get(SegmentArg.SHOW_CLUSTER, False):
 80        segments.append(
 81            SegmentData(
 82                contents=active_context["context"]["cluster"],
 83                highlight_groups=[HighlightGroup.KUBERNETES_CLUSTER.value],
 84                divider_highlight_group=HighlightGroup.KUBERNETES_NAMESPACE.value,
 85            )
 86        )
 87
 88    if (
 89        segment_args.get(SegmentArg.SHOW_DEFAULT_NAMESPACE, False)
 90        and "namespace" in active_context["context"]
 91    ):
 92        segments.extend(
 93            [
 94                SegmentData(
 95                    contents=" ",
 96                    highlight_groups=[HighlightGroup.KUBERNETES_DIVIDER.value],
 97                    divider_highlight_group=HighlightGroup.KUBERNETES_DIVIDER.value,
 98                ),
 99                SegmentData(
100                    contents=active_context["context"]["namespace"],
101                    highlight_groups=[HighlightGroup.KUBERNETES_CLUSTER.value],
102                    divider_highlight_group=HighlightGroup.KUBERNETES_NAMESPACE.value,
103                ),
104            ]
105        )
106
107    return [asdict(s) for s in segments]