Quickstart

Get up and running with OpenShift Python Wrapper in minutes

This guide will help you perform your first operations with the OpenShift Python Wrapper. You’ll learn how to connect to a cluster, create resources, and use the wrapper’s key features.

Prerequisites

1. Ensure Python 3.10+ is installed

   python --version
   # Should output Python 3.10 or higher
   
2. Install the library

   uv pip install openshift-python-wrapper
   
3. Configure cluster access

   Ensure you have a valid kubeconfig file or cluster credentials. Test your connection:

   kubectl cluster-info
   

Your First Resource

Let’s create a simple Namespace resource to understand the basic workflow.

Basic Example

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

# Get a client to connect to your cluster
client = get_client()

# Create a namespace
ns = Namespace(client=client, name="my-first-namespace")
ns.deploy()

# Verify it exists
print(f"Namespace exists: {ns.exists}")
print(f"Namespace status: {ns.status}")

# Clean up
ns.clean_up()

The context manager pattern (with statement) is recommended as it ensures automatic resource cleanup, even if an error occurs.

Working with Pods

Pods are fundamental workload resources. Here’s how to create and interact with them:

Create a Pod

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

client = get_client()

# Create a pod with an nginx container
with Pod(
    client=client,
    name="nginx-pod",
    namespace="default",
    containers=[{
        "name": "nginx",
        "image": "nginx:latest",
        "ports": [{"containerPort": 80}]
    }]
) as pod:
    # Wait for pod to be running
    pod.wait_for_status(status=Pod.Status.RUNNING, timeout=300)
    
    print(f"Pod {pod.name} is running on node: {pod.node.name}")
    
    # Get pod logs
    logs = pod.log()
    print(f"Pod logs:\n{logs}")

Working with Deployments

Deployments manage replica sets and pod lifecycles:

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

client = get_client()

with Deployment(
    client=client,
    name="nginx-deployment",
    namespace="default",
    replicas=3,
    selector={"matchLabels": {"app": "nginx"}},
    template={
        "metadata": {"labels": {"app": "nginx"}},
        "spec": {
            "containers": [{
                "name": "nginx",
                "image": "nginx:1.21",
                "ports": [{"containerPort": 80}]
            }]
        }
    }
) as deployment:
    deployment.deploy()
    
    # Wait for deployment to be ready
    deployment.wait_for_replicas(timeout=300)
    
    print(f"Deployment {deployment.name} is ready!")
    print(f"Available replicas: {deployment.instance.status.availableReplicas}")

Advanced Patterns

Using Resource Body

For complex resources, you can provide a complete body dictionary:

from ocp_resources.virtual_machine import VirtualMachine
from ocp_resources.resource import get_client

client = get_client()

vm = VirtualMachine(
    client=client,
    name="my-vm",
    namespace="default",
    body={
        "spec": {
            "runStrategy": "Halted",
            "template": {
                "spec": {
                    "domain": {
                        "devices": {
                            "disks": [{"name": "disk0", "disk": {"bus": "virtio"}}]
                        },
                        "resources": {"requests": {"memory": "64Mi"}}
                    },
                    "volumes": [{
                        "name": "disk0",
                        "containerDisk": {"image": "kubevirt/cirros-container-disk-demo"}
                    }]
                }
            }
        }
    }
)

vm.create()
vm.start()
vm.vmi.wait_until_running(timeout=180)

Creating Multiple Resources

Use ResourceList to create multiple similar resources efficiently:

from ocp_resources.namespace import Namespace
from ocp_resources.role import Role
from ocp_resources.resource import ResourceList, NamespacedResourceList, get_client

client = get_client()

# Create 3 namespaces: ns-1, ns-2, ns-3
with ResourceList(
    resource_class=Namespace,
    num_resources=3,
    client=client,
    name="ns"
) as namespaces:
    print(f"Created namespaces: {[ns.name for ns in namespaces]}")
    
    # Create a role in each namespace
    roles = NamespacedResourceList(
        client=client,
        resource_class=Role,
        name="viewer-role",
        namespaces=[ns.name for ns in namespaces],
        rules=[{
            "apiGroups": [""],
            "resources": ["pods", "services"],
            "verbs": ["get", "list", "watch"]
        }]
    )
    roles.deploy()
    
    print(f"Created roles in namespaces")
    roles.clean_up()

Resource with Different API Groups

Some resources exist in multiple API groups:

from ocp_resources.network import Network
from ocp_resources.resource import get_client

client = get_client()

# Default API group (config.openshift.io)
network_config = Network(client=client, name="cluster")

# Specific API group (operator.openshift.io)
network_operator = Network(
    client=client,
    name="cluster",
    api_group=Network.ApiGroup.OPERATOR_OPENSHIFT_IO
)

Schema Validation

The wrapper includes built-in schema validation to catch errors early:

Manual Validation

from ocp_resources.pod import Pod
from ocp_resources.exceptions import ValidationError

# Create a pod and validate before deploying
pod = Pod(
    name="nginx-pod",
    namespace="default",
    containers=[{"name": "nginx", "image": "nginx:latest"}]
)

try:
    pod.validate()
    print("Pod configuration is valid!")
except ValidationError as e:
    print(f"Validation failed: {e}")

Common Operations

Wait for Conditions

Wait for resources to reach desired states:

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

client = get_client()

pod = Pod(client=client, name="my-pod", namespace="default")

# Wait for specific status
pod.wait_for_status(status=Pod.Status.RUNNING, timeout=300)

# Wait for condition
pod.wait_for_condition(
    condition="Ready",
    status="True",
    timeout=300
)

Resource Discovery

Query and filter resources:

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

client = get_client()

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

# Filter by labels
for pod in Pod.get(client=client, label_selector="app=nginx,tier=frontend"):
    print(f"Found pod: {pod.name}")

# Get by field selector
for pod in Pod.get(client=client, field_selector="status.phase=Running"):
    print(f"Running pod: {pod.name}")

Resource Updates

Update existing resources:

from ocp_resources.config_map import ConfigMap
from ocp_resources.resource import get_client

client = get_client()

cm = ConfigMap(
    client=client,
    name="app-config",
    namespace="default",
    data={"key1": "value1"}
)
cm.deploy()

# Update the resource
cm.instance.data["key2"] = "value2"
cm.update()

print(f"ConfigMap updated: {cm.instance.data}")

Error Handling

Handle common scenarios gracefully:

from ocp_resources.namespace import Namespace
from ocp_resources.resource import get_client
from ocp_resources.exceptions import ResourceTeardownError
from timeout_sampler import TimeoutExpiredError
from kubernetes.dynamic.exceptions import NotFoundError

client = get_client()

try:
    ns = Namespace(client=client, name="my-namespace")
    ns.deploy()
    ns.wait_for_status(status=Namespace.Status.ACTIVE, timeout=120)
    
except TimeoutExpiredError:
    print("Resource did not reach desired state in time")
except NotFoundError:
    print("Resource not found on cluster")
except ResourceTeardownError:
    print("Failed to clean up resource")
finally:
    # Ensure cleanup
    if ns.exists:
        ns.clean_up()

Testing with Fake Client

Write tests without requiring a real cluster:

import pytest
from ocp_resources.namespace import Namespace
from ocp_resources.pod import Pod
from ocp_resources.resource import get_client

def test_pod_creation():
    """Test pod creation using fake client"""
    client = get_client(fake=True)
    
    with Namespace(client=client, name="test-ns") as ns:
        ns.deploy()
        assert ns.exists
        
        with Pod(
            client=client,
            name="test-pod",
            namespace=ns.name,
            containers=[{"name": "nginx", "image": "nginx:latest"}]
        ) as pod:
            pod.deploy()
            assert pod.exists
            assert pod.namespace == "test-ns"

if __name__ == "__main__":
    test_pod_creation()
    print("Test passed!")

Best Practices

Use Context Managers

Always prefer context managers (with statements) for automatic cleanup:

# Good
with Namespace(client=client, name="test") as ns:
    # Work with namespace
    pass  # Automatic cleanup

# Avoid
ns = Namespace(client=client, name="test")
ns.deploy()
# ... do work ...
ns.clean_up()  # Easy to forget!

Enable Schema Validation

Catch configuration errors early:

pod = Pod(
    client=client,
    name="my-pod",
    namespace="default",
    containers=[...],
    schema_validation_enabled=True
)

Wait for Resource Ready State

Don’t assume immediate availability:

pod.deploy()
pod.wait_for_status(status=Pod.Status.RUNNING, timeout=300)
# Now safe to interact with the pod

Use Appropriate Timeouts

Set realistic timeouts based on resource type:

# Pods might take a few minutes
pod.wait_for_status(status=Pod.Status.RUNNING, timeout=300)

# Deployments can take longer
deployment.wait_for_replicas(timeout=600)

Handle Cleanup Failures

Ensure resources are cleaned up even on errors:

from ocp_resources.exceptions import ResourceTeardownError

try:
    with Namespace(client=client, name="test") as ns:
        # ... work ...
        pass
except ResourceTeardownError as e:
    print(f"Warning: Cleanup failed: {e}")
    # Handle or log the error

Next Steps

Core Concepts

Learn about resources, namespaces, and context managers

API Reference

Explore all available resource classes and methods

Advanced Testing

See real-world examples and patterns

Fake Client

Learn how to test your code with the fake client

API Reference

Explore all available resource classes and methods

Class Generator

Auto-generate resource wrappers

Testing Guide

Learn how to test your code with the fake client