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.
Indicates if this resource should be deleted on cleanup
Path to YAML file for the resource
delete_timeout
int
default:"TIMEOUT_4MINUTES"
Timeout in seconds for delete operations
If True, perform a dry run without creating the resource
Node selector for pod scheduling
Node selector labels for pod scheduling
Path to kubeconfig file for connecting to remote cluster (deprecated, use client instead)
Dictionary with kubeconfig configuration
Context name for connecting to remote cluster (deprecated, use client instead)
Resource labels to add to metadata
Resource annotations to add to metadata
Resource API group; will overwrite API group definition in resource class
Hash resource content based on resource keys_to_hash property (e.g., Secret data)
Whether to check if the resource exists when initializing, raise if not
Dictionary representing the resource object
Waits for the resource to be created when using deploy()
schema_validation_enabled
Enable automatic schema validation for this instance. Set to True to validate on create/update operations.
Class Attributes
API group for the resource (e.g., “apps”, “batch”)
API version (e.g., “v1”, “v1beta1”)
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
True to wait for resource status after creation
exceptions_dict
dict[type[Exception], list[str]]
Dictionary of exceptions to retry on
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 for resource deletion to complete
timeout
int
default:"TIMEOUT_4MINUTES"
Timeout in seconds to wait for resource deletion
Optional request body for deletion
update
Update the resource with a resource dictionary (partial update/patch).
def update(self, resource_dict: dict[str, Any]) -> None
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
Complete resource dictionary to replace with
deploy
Deploy the resource using context manager pattern.
def deploy(self, wait: bool = False) -> Self
Wait for resource to be ready after deployment
clean_up
Clean up (delete) the resource.
def clean_up(self, wait: bool = True, timeout: int | None = None) -> bool
Wait for resource deletion
Timeout in seconds, defaults to delete_timeout
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
Time to wait between retries
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
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
Expected status (e.g., “Running”, “Succeeded”)
timeout
int
default:"TIMEOUT_4MINUTES"
Time to wait for the resource
Status which should stop the wait and fail (defaults to “Failed”)
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 type to query (e.g., “Ready”, “Available”)
Expected condition status (e.g., “True”, “False”)
Interval between each retry
Expected condition reason
Expected text in condition message
Condition which should stop the wait and fail
Status of the stop condition
validate
Validate the resource against its OpenAPI schema.
def validate(self) -> None
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]
Resource kind in lowercase for disambiguation
If True, return raw ResourceInstance objects
Properties
exists
@property
def exists(self) -> ResourceInstance | None
instance
@property
def instance(self) -> ResourceInstance
status
@property
def status(self) -> str
labels
@property
def labels(self) -> ResourceField
api
@property
def api(self) -> ResourceInstance
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
