Resource and NamespacedResource API

from ocp_resources.resource import (
    Resource,
    NamespacedResource,
    ResourceEditor,
    ResourceList,
    NamespacedResourceList,
    get_client,
)

from ocp_resources.resource import get_client

client = get_client(config_file="~/.kube/config")

# Default kubeconfig
client = get_client()

# With explicit config file and context
client = get_client(config_file="/path/to/kubeconfig", context="my-cluster")

# Token-based auth
client = get_client(host="https://api.cluster.example.com:6443", token="sha256~abc123")

# Basic auth
client = get_client(host="https://api.cluster.example.com:6443", username="admin", password="secret")

# Fake client for unit testing
client = get_client(fake=True)

from ocp_resources.resource import Resource

Resource(
    client=client,
    name="my-resource",
)

from ocp_resources.namespace import Namespace

# Basic creation
ns = Namespace(client=client, name="my-namespace")

# From a YAML file
ns = Namespace(client=client, yaml_file="namespace.yaml")

# From a dict
ns = Namespace(client=client, kind_dict={"apiVersion": "v1", "kind": "Namespace", "metadata": {"name": "test"}})

# With labels and annotations
ns = Namespace(client=client, name="my-ns", label={"env": "test"}, annotations={"owner": "team-a"})

# Verify it already exists
ns = Namespace(client=client, name="default", ensure_exists=True)

from ocp_resources.namespace import Namespace

with Namespace(client=client, name="temp-ns") as ns:
    # Resource is created on __enter__
    print(ns.name)
# Resource is deleted on __exit__ (if teardown=True)

def create(
    self,
    wait: bool = False,
    exceptions_dict: dict[type[Exception], list[str]] = ...,
) -> ResourceInstance | None

ns = Namespace(client=client, name="my-ns")
ns.create(wait=True)

def delete(
    self,
    wait: bool = False,
    timeout: int = 240,
    body: dict[str, Any] | None = None,
) -> bool

ns.delete(wait=True, timeout=120)

def update(self, resource_dict: dict[str, Any]) -> None

ns.update(resource_dict={"metadata": {"labels": {"env": "staging"}}})

def update_replace(self, resource_dict: dict[str, Any]) -> None

full_dict = ns.instance.to_dict()
full_dict["metadata"]["labels"] = {"new-label": "only"}
ns.update_replace(resource_dict=full_dict)

def deploy(self, wait: bool = False) -> Self

def clean_up(self, wait: bool = True, timeout: int | None = None) -> bool

@property
def exists(self) -> ResourceInstance | None

if ns.exists:
    print("Namespace exists")

@property
def instance(self) -> ResourceInstance

resource_version = ns.instance.metadata.resourceVersion

@property
def status(self) -> str

print(ns.status)  # "Active"

@property
def labels(self) -> ResourceField

for key, value in ns.labels.items():
    print(f"{key}={value}")

@ClassProperty
def kind(cls) -> str | None

from ocp_resources.namespace import Namespace
print(Namespace.kind)  # "Namespace"

@property
def api(self) -> ResourceInstance

def full_api(self, **kwargs: Any) -> ResourceInstance

@classmethod
def get(
    cls,
    client: DynamicClient | None = None,
    dyn_client: DynamicClient | None = None,
    config_file: str = "",
    singular_name: str = "",
    exceptions_dict: dict[type[Exception], list[str]] = ...,
    raw: bool = False,
    context: str | None = None,
    *args: Any,
    **kwargs: Any,
) -> Generator[Any, None, None]

from ocp_resources.namespace import Namespace

for ns in Namespace.get(client=client):
    print(ns.name)

# With label selector
for ns in Namespace.get(client=client, label_selector="env=production"):
    print(ns.name)

# Raw mode
for raw_ns in Namespace.get(client=client, raw=True):
    print(raw_ns.metadata.name, raw_ns.status.phase)

@staticmethod
def get_all_cluster_resources(
    client: DynamicClient | None = None,
    config_file: str = "",
    context: str | None = None,
    config_dict: dict[str, Any] | None = None,
    *args: Any,
    **kwargs: Any,
) -> Generator[ResourceField, None, None]

for resource in Resource.get_all_cluster_resources(client=client, label_selector="app=myapp"):
    print(f"{resource.kind}: {resource.metadata.name}")

def wait(self, timeout: int = 240, sleep: int = 1) -> None

def wait_deleted(self, timeout: int = 240) -> bool

def wait_for_status(
    self,
    status: str,
    timeout: int = 240,
    stop_status: str | None = None,
    sleep: int = 1,
    exceptions_dict: dict[type[Exception], list[str]] = ...,
) -> None

from ocp_resources.pod import Pod

pod.wait_for_status(status=Pod.Status.RUNNING, timeout=120)

def wait_for_condition(
    self,
    condition: str,
    status: str,
    timeout: int = 300,
    sleep_time: int = 1,
    reason: str | None = None,
    message: str = "",
    stop_condition: str | None = None,
    stop_status: str = "True",
) -> None

ns.wait_for_condition(
    condition="Ready",
    status="True",
    timeout=60,
)

def wait_for_conditions(self) -> None

def to_dict(self) -> None

def to_yaml(self) -> str

print(ns.to_yaml())

def watcher(
    self,
    timeout: int,
    resource_version: str = "",
) -> Generator[dict[str, Any], None, None]

for event in ns.watcher(timeout=30):
    print(event["type"], event["object"].metadata.name)

def events(
    self,
    name: str = "",
    label_selector: str = "",
    field_selector: str = "",
    resource_version: str = "",
    timeout: int = 240,
) -> Generator[Any, Any, None]

for event in pod.events(field_selector="type==Warning", timeout=10):
    print(event.object)

def validate(self) -> None

pod = Pod(client=client, name="test", namespace="default")
pod.to_dict()
pod.validate()  # Raises ValidationError on invalid spec

@classmethod
def validate_dict(cls, resource_dict: dict[str, Any]) -> None

from ocp_resources.pod import Pod

Pod.validate_dict({
    "apiVersion": "v1",
    "kind": "Pod",
    "metadata": {"name": "test"},
    "spec": {"containers": [{"name": "web", "image": "nginx"}]},
})

def api_request(
    self,
    method: str,
    action: str,
    url: str,
    retry_params: dict[str, int] | None = None,
    **params: Any,
) -> dict[str, Any]

@staticmethod
def retry_cluster_exceptions(
    func: Callable,
    exceptions_dict: dict[type[Exception], list[str]] = DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
    timeout: int = 10,
    sleep_time: int = 1,
    **kwargs: Any,
) -> Any

def get_condition_message(
    self,
    condition_type: str,
    condition_status: str = "",
) -> str

def hash_resource_dict(self, resource_dict: dict[Any, Any]) -> dict[Any, Any]

@property
def keys_to_hash(self) -> list[str]

# Example override in a Secret subclass:
@property
def keys_to_hash(self):
    return ["data", "stringData"]

pod.wait_for_status(status=Pod.Status.RUNNING)

ns.wait_for_condition(
    condition=Resource.Condition.READY,
    status=Resource.Condition.Status.TRUE,
)

from ocp_resources.resource import NamespacedResource

NamespacedResource(
    client=client,
    name="my-pod",
    namespace="my-namespace",
)

from ocp_resources.pod import Pod

pod = Pod(
    client=client,
    name="nginx",
    namespace="default",
    label={"app": "web"},
)

@property
def instance(self) -> ResourceInstance

def to_dict(self) -> None

from ocp_resources.pod import Pod

for pod in Pod.get(client=client, namespace="default"):
    print(f"{pod.namespace}/{pod.name}")

# With label selector
for pod in Pod.get(client=client, namespace="default", label_selector="app=web"):
    print(pod.name)

from ocp_resources.resource import ResourceEditor

ResourceEditor(
    patches: dict[Any, Any],
    action: str = "update",
    user_backups: dict[Any, Any] | None = None,
)

from ocp_resources.resource import ResourceEditor
from ocp_resources.namespace import Namespace

ns = Namespace(client=client, name="my-ns", ensure_exists=True)

with ResourceEditor(
    patches={ns: {"metadata": {"labels": {"temporary": "true"}}}}
) as editor:
    # Labels are patched
    assert ns.instance.metadata.labels.temporary == "true"
# Labels are restored to original values

from ocp_resources.resource import ResourceList

ResourceList(
    resource_class: type[Resource],
    num_resources: int,
    client: DynamicClient,
    **kwargs: Any,
)

from ocp_resources.namespace import Namespace
from ocp_resources.resource import ResourceList

namespaces = ResourceList(
    resource_class=Namespace,
    num_resources=3,
    client=client,
    name="test-ns",
)

with namespaces:
    # Creates test-ns-1, test-ns-2, test-ns-3
    for ns in namespaces:
        print(ns.name)
# All three namespaces are deleted

from ocp_resources.resource import NamespacedResourceList

NamespacedResourceList(
    resource_class: type[NamespacedResource],
    namespaces: ResourceList,
    client: DynamicClient,
    **kwargs: Any,
)

from ocp_resources.namespace import Namespace
from ocp_resources.pod import Pod
from ocp_resources.resource import ResourceList, NamespacedResourceList

namespaces = ResourceList(
    resource_class=Namespace, num_resources=2, client=client, name="test-ns"
)

pods = NamespacedResourceList(
    resource_class=Pod,
    namespaces=namespaces,
    client=client,
    name="nginx",
)

with namespaces:
    with pods:
        # Creates nginx in test-ns-1 and test-ns-2
        for pod in pods:
            print(f"{pod.namespace}/{pod.name}")

from ocp_resources.resource import KubeAPIVersion

KubeAPIVersion(vstring: str)

assert KubeAPIVersion("v1") > KubeAPIVersion("v1beta1")
assert KubeAPIVersion("v1beta1") > KubeAPIVersion("v1alpha1")
assert KubeAPIVersion("v1") == KubeAPIVersion("v1")

from ocp_resources.exceptions import (
    MissingRequiredArgumentError,
    ResourceTeardownError,
    ValidationError,
    ConditionError,
)

from ocp_resources.utils.constants import (
    TIMEOUT_1SEC,      # 1
    TIMEOUT_5SEC,      # 5
    TIMEOUT_10SEC,     # 10
    TIMEOUT_30SEC,     # 30
    TIMEOUT_1MINUTE,   # 60
    TIMEOUT_2MINUTES,  # 120
    TIMEOUT_4MINUTES,  # 240
    TIMEOUT_10MINUTES, # 600
)

from ocp_resources.resource import replace_key_with_hashed_value

replace_key_with_hashed_value(resource_dict: dict, key_name: str) -> dict

data = {"spec": {"secret": {"token": "sensitive"}}}
hashed = replace_key_with_hashed_value(data, "spec>secret>token")
# {"spec": {"secret": {"token": "*******"}}}

# With list path
data = {"spec": {"containers": [{"env": [{"value": "secret"}]}]}}
hashed = replace_key_with_hashed_value(data, "spec>containers[]>env[]>value")