# Path: libcloudforensics/errors.py
# class LCFError(Exception):
# class CredentialsConfigurationError(LCFError):
# class InvalidFileFormatError(LCFError):
# class InvalidNameError(LCFError):
# class ResourceNotFoundError(LCFError):
# class ResourceCreationError(LCFError):
# class ResourceDeletionError(LCFError):
# class InstanceStateChangeError(LCFError):
# class ServiceAccountRemovalError(LCFError):
# class InstanceProfileCreationError(LCFError):
# class OperationFailedError(LCFError):
# class TransferCreationError(LCFError):
# class TransferExecutionError(LCFError):
# class ResourceAlreadyExistsError(LCFError):
#   def __init__(self,
#                message: str,
#                name: str) -> None:
# 
# Path: libcloudforensics/providers/kubernetes/base.py
# class K8sClient(metaclass=abc.ABCMeta):
# class K8sResource(K8sClient, metaclass=abc.ABCMeta):
# class K8sNamespacedResource(K8sResource, metaclass=abc.ABCMeta):
# class K8sNode(K8sResource):
# class K8sWorkload(K8sNamespacedResource):
# class K8sPod(K8sWorkload):
#   T = TypeVar('T')
#   def __init__(self, api_client: client.ApiClient) -> None:
#   def _Api(self, api_class: Callable[[client.ApiClient], T]) -> T:
#   def __init__(self, api_client: client.ApiClient, name: str) -> None:
#   def Read(self) -> object:
#   def __init__(
#       self, api_client: client.ApiClient, name: str, namespace: str) -> None:
#   def Delete(self, cascade: bool = True) -> None:
#   def Read(self) -> client.V1Node:
#   def Cordon(self) -> None:
#   def Drain(self, pod_filter: Callable[['K8sPod'], bool]) -> None:
#   def ListPods(self, namespace: Optional[str] = None) -> List['K8sPod']:
#   def ExternalIps(self) -> List[str]:
#   def InternalIps(self) -> List[str]:
#   def GetCoveredPods(self) -> List['K8sPod']:
#   def GetCoveredNodes(self) -> List[K8sNode]:
#   def IsCoveringPod(self, pod: 'K8sPod') -> bool:
#   def gcp_protopayload_methodname(self) -> str:
#   def GcpClusterLogsQuerySupplement(self) -> str:
#   def GcpContainerLogsQuerySupplement(self) -> str:
#   def OrphanPods(self) -> None:
#   def gcp_protopayload_methodname(self) -> str:
#   def GcpContainerLogsQuerySupplement(self) -> str:
#   def OrphanPods(self) -> None:
#   def GetCoveredPods(self) -> List['K8sPod']:
#   def IsCoveringPod(self, pod: 'K8sPod') -> bool:
#   def Read(self) -> client.V1Pod:
#   def GetNode(self) -> K8sNode:
#   def ListContainers(self) -> List[container.K8sContainer]:
#   def ListVolumes(self) -> List[volume.K8sVolume]:
#   def GetLabels(self) -> Dict[str, str]:
#   def Delete(self, cascade: bool = True) -> None:
#   def AddLabels(self, labels: Dict[str, str]) -> None:
# 
# Path: libcloudforensics/providers/kubernetes/selector.py
# class K8sSelector:
#   class Component(abc.ABC):
#   class LabelComponent(Component, metaclass=abc.ABCMeta):
#   class FieldComponent(Component, metaclass=abc.ABCMeta):
#   class Name(FieldComponent):
#   class Node(FieldComponent):
#   class Running(FieldComponent):
#   class Label(LabelComponent):
#     def ToString(self) -> str:
#     def keyword(self) -> str:
#     def keyword(self) -> str:
#     def keyword(self) -> str:
#     def __init__(self, name: str) -> None:
#     def ToString(self) -> str:
#     def __init__(self, node: str) -> None:
#     def ToString(self) -> str:
#     def ToString(self) -> str:
#     def __init__(self, key: str, value: str) -> None:
#     def ToString(self) -> str:
#   def __init__(self, *selectors: Component) -> None:
#   def ToKeywords(self) -> Dict[str, str]:
#   def FromLabelsDict(cls, labels: Dict[str, str]) -> 'K8sSelector':


import abc
from typing import List, Dict, Union
from kubernetes import client
from libcloudforensics import errors
from libcloudforensics.providers.kubernetes import base
from libcloudforensics.providers.kubernetes import selector

  @abc.abstractmethod
  def _PodMatchLabels(self) -> Dict[str, str]:
    """Gets the key-value pairs that pods belonging to this workload would have.

    Returns:
      Dict[str, str]: The label key-value pairs of this workload's pods.
    """

  @abc.abstractmethod
  def Read(self) -> Union[client.V1Deployment, client.V1ReplicaSet,]:
    """Override of abstract method."""  # Narrows down type hint

  def MatchLabels(self) -> Dict[str, str]:
    """Gets the label key-value pairs in the matchLabels field.

    Returns:
      Dict[str, str]: The labels in the matchLabels field.

    Raises:
      NotImplementedError: If matchExpressions exist, in which case using the
          matchLabels will be inaccurate.
    """
    read = self.Read()

    if read.spec.selector.match_expressions:
      raise NotImplementedError(
          'matchExpressions exist, meaning using '
          'matchLabels will be inaccurate.')


# Path: libcloudforensics/errors.py
# class LCFError(Exception):
# class CredentialsConfigurationError(LCFError):
# class InvalidFileFormatError(LCFError):
# class InvalidNameError(LCFError):
# class ResourceNotFoundError(LCFError):
# class ResourceCreationError(LCFError):
# class ResourceDeletionError(LCFError):
# class InstanceStateChangeError(LCFError):
# class ServiceAccountRemovalError(LCFError):
# class InstanceProfileCreationError(LCFError):
# class OperationFailedError(LCFError):
# class TransferCreationError(LCFError):
# class TransferExecutionError(LCFError):
# class ResourceAlreadyExistsError(LCFError):
#   def __init__(self,
#                message: str,
#                name: str) -> None:
# 
# Path: libcloudforensics/providers/kubernetes/base.py
# class K8sClient(metaclass=abc.ABCMeta):
# class K8sResource(K8sClient, metaclass=abc.ABCMeta):
# class K8sNamespacedResource(K8sResource, metaclass=abc.ABCMeta):
# class K8sNode(K8sResource):
# class K8sWorkload(K8sNamespacedResource):
# class K8sPod(K8sWorkload):
#   T = TypeVar('T')
#   def __init__(self, api_client: client.ApiClient) -> None:
#   def _Api(self, api_class: Callable[[client.ApiClient], T]) -> T:
#   def __init__(self, api_client: client.ApiClient, name: str) -> None:
#   def Read(self) -> object:
#   def __init__(
#       self, api_client: client.ApiClient, name: str, namespace: str) -> None:
#   def Delete(self, cascade: bool = True) -> None:
#   def Read(self) -> client.V1Node:
#   def Cordon(self) -> None:
#   def Drain(self, pod_filter: Callable[['K8sPod'], bool]) -> None:
#   def ListPods(self, namespace: Optional[str] = None) -> List['K8sPod']:
#   def ExternalIps(self) -> List[str]:
#   def InternalIps(self) -> List[str]:
#   def GetCoveredPods(self) -> List['K8sPod']:
#   def GetCoveredNodes(self) -> List[K8sNode]:
#   def IsCoveringPod(self, pod: 'K8sPod') -> bool:
#   def gcp_protopayload_methodname(self) -> str:
#   def GcpClusterLogsQuerySupplement(self) -> str:
#   def GcpContainerLogsQuerySupplement(self) -> str:
#   def OrphanPods(self) -> None:
#   def gcp_protopayload_methodname(self) -> str:
#   def GcpContainerLogsQuerySupplement(self) -> str:
#   def OrphanPods(self) -> None:
#   def GetCoveredPods(self) -> List['K8sPod']:
#   def IsCoveringPod(self, pod: 'K8sPod') -> bool:
#   def Read(self) -> client.V1Pod:
#   def GetNode(self) -> K8sNode:
#   def ListContainers(self) -> List[container.K8sContainer]:
#   def ListVolumes(self) -> List[volume.K8sVolume]:
#   def GetLabels(self) -> Dict[str, str]:
#   def Delete(self, cascade: bool = True) -> None:
#   def AddLabels(self, labels: Dict[str, str]) -> None:
# 
# Path: libcloudforensics/providers/kubernetes/selector.py
# class K8sSelector:
#   class Component(abc.ABC):
#   class LabelComponent(Component, metaclass=abc.ABCMeta):
#   class FieldComponent(Component, metaclass=abc.ABCMeta):
#   class Name(FieldComponent):
#   class Node(FieldComponent):
#   class Running(FieldComponent):
#   class Label(LabelComponent):
#     def ToString(self) -> str:
#     def keyword(self) -> str:
#     def keyword(self) -> str:
#     def keyword(self) -> str:
#     def __init__(self, name: str) -> None:
#     def ToString(self) -> str:
#     def __init__(self, node: str) -> None:
#     def ToString(self) -> str:
#     def ToString(self) -> str:
#     def __init__(self, key: str, value: str) -> None:
#     def ToString(self) -> str:
#   def __init__(self, *selectors: Component) -> None:
#   def ToKeywords(self) -> Dict[str, str]:
#   def FromLabelsDict(cls, labels: Dict[str, str]) -> 'K8sSelector':


# Path: libcloudforensics/providers/kubernetes/workloads.py
import abc
from typing import List, Dict, Union
from kubernetes import client
from libcloudforensics import errors
from libcloudforensics.providers.kubernetes import base
from libcloudforensics.providers.kubernetes import selector


  @abc.abstractmethod
  def _PodMatchLabels(self) -> Dict[str, str]:
    """Gets the key-value pairs that pods belonging to this workload would have.

    Returns:
      Dict[str, str]: The label key-value pairs of this workload's pods.
    """

  @abc.abstractmethod
  def Read(self) -> Union[client.V1Deployment, client.V1ReplicaSet,]:
    """Override of abstract method."""  # Narrows down type hint

  def MatchLabels(self) -> Dict[str, str]:
    """Gets the label key-value pairs in the matchLabels field.

    Returns:
      Dict[str, str]: The labels in the matchLabels field.

    Raises:
      NotImplementedError: If matchExpressions exist, in which case using the
          matchLabels will be inaccurate.
    """
    read = self.Read()

    if read.spec.selector.match_expressions:
      raise NotImplementedError(
          'matchExpressions exist, meaning using '
          'matchLabels will be inaccurate.')


    match_labels = read.spec.selector.match_labels  # type: Dict[str, str]