Resource

Base class for all Kubernetes/OpenShift resources

Overview

The Resource class is the base class for all API resources in the OpenShift Python Wrapper. It provides common functionality for CRUD operations, resource management, schema validation, and interaction with Kubernetes/OpenShift clusters.

All resource classes inherit from this base class, either directly or through NamespacedResource.

Class Definition

from ocp_resources.resource import Resource

Constructor

client · DynamicClient | None · required

Dynamic client for connecting to a remote cluster. Will become mandatory in the next major release.

name · str | None

Resource name

teardown · bool · default: "True"

Indicates if this resource should be deleted on cleanup

yaml_file · str | None

Path to YAML file for the resource

delete_timeout · int · default: "TIMEOUT_4MINUTES"

Timeout in seconds for delete operations

dry_run · bool · default: "False"

If True, perform a dry run without creating the resource

node_selector · dict[str, Any] | None

Node selector for pod scheduling

node_selector_labels · dict[str, str] | None

Node selector labels for pod scheduling

config_file · str | None

Path to kubeconfig file for connecting to remote cluster (deprecated, use client instead)

config_dict · dict[str, Any] | None

Dictionary with kubeconfig configuration

context · str | None

Context name for connecting to remote cluster (deprecated, use client instead)

label · dict[str, str] | None

Resource labels to add to metadata

annotations · dict[str, str] | None

Resource annotations to add to metadata

api_group · str · default: "''"

Resource API group; will overwrite API group definition in resource class

hash_log_data · bool · default: "True"

Hash resource content based on resource keys_to_hash property (e.g., Secret data)

ensure_exists · bool · default: "False"

Whether to check if the resource exists when initializing, raise if not

kind_dict · dict[Any, Any] | None

Dictionary representing the resource object

wait_for_resource · bool · default: "False"

Waits for the resource to be created when using deploy()

schema_validation_enabled · bool · default: "False"

Enable automatic schema validation for this instance. Set to True to validate on create/update operations.

Class Attributes

api_group · str

API group for the resource (e.g., “apps”, “batch”)

api_version · str

API version (e.g., “v1”, “v1beta1”)

singular_name · str

Singular resource name for API calls

timeout_seconds · int · default: "TIMEOUT_1MINUTE"

Default timeout for API operations

Methods

create

Create the resource in the cluster.

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

wait · bool · default: "False"

True to wait for resource status after creation

exceptions_dict · dict[type[Exception], list[str]]

Dictionary of exceptions to retry on

Returns: Created resource instance or None if create failed

delete

Delete the resource from the cluster.

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

wait · bool · default: "False"

Wait for resource deletion to complete

timeout · int · default: "TIMEOUT_4MINUTES"

Timeout in seconds to wait for resource deletion

body · dict[str, Any] | None

Optional request body for deletion

Returns: True if resource was deleted, False otherwise

update

Update the resource with a resource dictionary (partial update/patch).

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

resource_dict · dict[str, Any] · required

Resource dictionary with fields to update

update_replace

Replace the resource metadata completely. Use this to remove existing fields.

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

resource_dict · dict[str, Any] · required

Complete resource dictionary to replace with

deploy

Deploy the resource using context manager pattern.

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

wait · bool · default: "False"

Wait for resource to be ready after deployment

Returns: Self for method chaining

clean_up

Clean up (delete) the resource.

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

wait · bool · default: "True"

Wait for resource deletion

timeout · int | None

Timeout in seconds, defaults to delete_timeout

Returns: True if resource was deleted, False otherwise

wait

Wait for the resource to exist.

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

timeout · int · default: "TIMEOUT_4MINUTES"

Time to wait for the resource

sleep · int · default: "1"

Time to wait between retries

Raises: TimeoutExpiredError if resource does not exist within timeout

wait_deleted

Wait until the resource is deleted.

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

timeout · int · default: "TIMEOUT_4MINUTES"

Time to wait for the resource

Returns: True if deleted, False if timeout expired

wait_for_status

Wait for the resource to reach a specific status.

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

status · str · required

Expected status (e.g., “Running”, “Succeeded”)

timeout · int · default: "TIMEOUT_4MINUTES"

Time to wait for the resource

stop_status · str | None

Status which should stop the wait and fail (defaults to “Failed”)

sleep · int · default: "1"

Time between status checks

wait_for_condition

Wait for a resource condition to reach the desired status.

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

condition · str · required

Condition type to query (e.g., “Ready”, “Available”)

status · str · required

Expected condition status (e.g., “True”, “False”)

timeout · int · default: "300"

Time to wait in seconds

sleep_time · int · default: "1"

Interval between each retry

reason · str | None

Expected condition reason

message · str · default: "''"

Expected text in condition message

stop_condition · str | None

Condition which should stop the wait and fail

stop_status · str · default: "'True'"

Status of the stop condition

validate

Validate the resource against its OpenAPI schema.

def validate(self) -> None

Raises: ValidationError if the resource is invalid according to the schema

get (class method)

Get resources from the cluster.

@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]] = DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
    raw: bool = False,
    context: str | None = None,
    *args: Any,
    **kwargs: Any,
) -> Generator[Any, None, None]

client · DynamicClient | None

Kubernetes client

singular_name · str · default: "''"

Resource kind in lowercase for disambiguation

raw · bool · default: "False"

If True, return raw ResourceInstance objects

Returns: Generator of resource objects

Properties

exists

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

Returns the resource instance if it exists on the server, None otherwise.

instance

@property
def instance(self) -> ResourceInstance

Get the current resource instance from the cluster.

status

@property
def status(self) -> str

Get resource status phase (e.g., “Running”, “Pending”, “Failed”).

labels

@property
def labels(self) -> ResourceField

Get resource labels.

api

@property
def api(self) -> ResourceInstance

Get the API resource instance for making API calls.

Context Manager Usage

The Resource class can be used as a context manager for automatic cleanup:

from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

# Automatically cleans up the pod when exiting the context
with Pod(
    client=client,
    name="my-pod",
    namespace="default",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
) as pod:
    print(f"Pod {pod.name} is running")
    # Pod is automatically deleted when exiting this block

Examples

Basic Resource Creation

from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client

client = get_client()

# Create a namespace
ns = Namespace(client=client, name="my-namespace")
ns.create(wait=True)

print(f"Namespace {ns.name} created")

# Clean up
ns.delete(wait=True)

Using YAML File

from ocp_resources.deployment import Deployment
from ocp_resources.resource import get_client

client = get_client()

deploy = Deployment(
    client=client,
    yaml_file="/path/to/deployment.yaml"
)
deploy.create()

Waiting for Conditions

from ocp_resources.deployment import Deployment
from ocp_resources.resource import get_client

client = get_client()

deploy = Deployment(client=client, name="my-app", namespace="default")

# Wait for the deployment to be available
deploy.wait_for_condition(
    condition="Available",
    status="True",
    timeout=300
)

Schema Validation

from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

pod = Pod(
    client=client,
    name="validated-pod",
    namespace="default",
    containers=[{"name": "nginx", "image": "nginx:latest"}],
    schema_validation_enabled=True  # Enable automatic validation
)

# Will validate against OpenAPI schema before creating
pod.create()

Listing Resources

from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

client = get_client()

# Get all pods in a namespace
for pod in Pod.get(client=client, namespace="default"):
    print(f"Pod: {pod.name}, Status: {pod.status}")

See Also

• NamespacedResource - Base class for namespaced resources
• Pod - Pod resource implementation
• Deployment - Deployment resource implementation