Pod

Pod resource for running containers in Kubernetes/OpenShift

Overview

The Pod class represents a Kubernetes Pod - a collection of one or more containers that run on a host. Pods are the smallest deployable units in Kubernetes and are scheduled onto nodes by the scheduler.

Class Definition

from ocp_resources.pod import Pod

API Version: v1

Constructor

Pod accepts all parameters from NamespacedResource, plus:

containers · list[Any] · required

List of containers belonging to the pod. Must contain at least one container. Each container should be a dictionary with keys like name, image, command, etc.

active_deadline_seconds · int | None

Optional duration in seconds the pod may be active before the system tries to mark it failed and kill associated containers

affinity · dict[str, Any] | None

Affinity scheduling rules for the pod

automount_service_account_token · bool | None

Whether a service account token should be automatically mounted

dns_config · dict[str, Any] | None

DNS parameters for the pod

dns_policy · str | None

DNS policy for the pod. Valid values: ClusterFirst, ClusterFirstWithHostNet, Default, None

enable_service_links · bool | None

Whether information about services should be injected into pod’s environment variables

ephemeral_containers · list[Any] | None

List of ephemeral containers for debugging

host_aliases · list[Any] | None

Optional list of hosts and IPs to inject into the pod’s hosts file

host_ipc · bool | None

Use the host’s IPC namespace

host_network · bool | None

Use the host’s network namespace

host_pid · bool | None

Use the host’s PID namespace

host_users · bool | None

Use the host’s user namespace

hostname · str | None

Hostname of the pod

image_pull_secrets · list[Any] | None

List of references to secrets for pulling images

init_containers · list[Any] | None

List of initialization containers executed before main containers

node_name · str | None

NodeName indicates which node this pod is scheduled on

node_selector · dict[str, Any] | None

Selector which must match a node’s labels for the pod to be scheduled on that node

os · dict[str, Any] | None

OS parameters of the pod

overhead · dict[str, Any] | None

Resource overhead associated with running a pod

preemption_policy · str | None

Policy for preempting pods with lower priority. Values: Never, PreemptLowerPriority

priority · int | None

Priority value for the pod

priority_class_name · str | None

Priority class name (e.g., system-node-critical, system-cluster-critical)

readiness_gates · list[Any] | None

Readiness gates for pod readiness evaluation

resource_claims · list[Any] | None

ResourceClaims that must be allocated before the pod starts

resources · dict[str, Any] | None

Compute resource requirements

restart_policy · str | None

Restart policy for containers. Values: Always, OnFailure, Never

runtime_class_name · str | None

RuntimeClass name to use for running this pod

scheduler_name · str | None

Name of the scheduler to dispatch this pod

scheduling_gates · list[Any] | None

SchedulingGates that block pod scheduling

security_context · dict[str, Any] | None

Pod-level security attributes

service_account · str | None

Deprecated alias for service_account_name

service_account_name · str | None

Name of the ServiceAccount to use

set_hostname_as_fqdn · bool | None

If true, the pod’s hostname will be configured as the pod’s FQDN

share_process_namespace · bool | None

Share a single process namespace between all containers

subdomain · str | None

Subdomain for the pod’s fully qualified hostname

termination_grace_period_seconds · int | None

Duration in seconds the pod needs to terminate gracefully

tolerations · list[Any] | None

Pod’s tolerations

topology_spread_constraints · list[Any] | None

How pods should spread across topology domains

volumes · list[Any] | None

List of volumes that can be mounted by containers

Methods

execute

Execute a command in the pod.

def execute(
    self,
    command: list[str],
    timeout: int = 60,
    container: str = "",
    ignore_rc: bool = False
) -> str

command · list[str] · required

Command to run in the pod

timeout · int · default: "60"

Time to wait for the command to complete

container · str · default: "''"

Container name where to exec the command. If empty, uses the first container.

ignore_rc · bool · default: "False"

If True, ignore error return code and return output anyway

Returns: Command output as string

Raises: ExecOnPodError if the command failed

log

Get pod logs.

def log(self, **kwargs: Any) -> str

Returns: Pod logs as string

Accepts all kwargs supported by Kubernetes API for reading pod logs, such as:

• container: Container name
• follow: Follow the log stream
• previous: Get logs from previous container instance
• since_seconds: Relative time in seconds
• tail_lines: Number of lines from the end
• timestamps: Include timestamps

Properties

node

@property
def node(self) -> Node

Get the Node object where the pod is running.

Returns: Node object

ip

@property
def ip(self) -> str

Get the pod’s IP address.

Returns: Pod IP as string

Inherited Methods

Pod inherits all methods from NamespacedResource and Resource.

Examples

Creating a Simple Pod

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",
    containers=[{
        "name": "nginx",
        "image": "nginx:latest",
        "ports": [{"containerPort": 80}]
    }]
)
pod.create(wait=True)

print(f"Pod {pod.name} created with IP: {pod.ip}")

Creating a Pod with Multiple Containers

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

client = get_client()

pod = Pod(
    client=client,
    name="multi-container-pod",
    namespace="default",
    containers=[
        {
            "name": "nginx",
            "image": "nginx:latest",
            "ports": [{"containerPort": 80}]
        },
        {
            "name": "sidecar",
            "image": "busybox:latest",
            "command": ["sh", "-c", "while true; do sleep 30; done"]
        }
    ]
)
pod.create()

Executing Commands in a Pod

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"
)

# Execute a command
output = pod.execute(command=["ls", "-la", "/"])
print(output)

# Execute in specific container
output = pod.execute(
    command=["cat", "/etc/hostname"],
    container="nginx"
)
print(output)

Getting Pod Logs

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")

# Get all logs
logs = pod.log()
print(logs)

# Get logs from specific container
logs = pod.log(container="nginx")

# Get last 100 lines with timestamps
logs = pod.log(tail_lines=100, timestamps=True)

Pod with Resource Limits

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

client = get_client()

pod = Pod(
    client=client,
    name="limited-pod",
    namespace="default",
    containers=[{
        "name": "nginx",
        "image": "nginx:latest",
        "resources": {
            "requests": {
                "memory": "64Mi",
                "cpu": "250m"
            },
            "limits": {
                "memory": "128Mi",
                "cpu": "500m"
            }
        }
    }]
)
pod.create()

Pod with Volume Mounts

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

client = get_client()

pod = Pod(
    client=client,
    name="pod-with-volume",
    namespace="default",
    containers=[{
        "name": "nginx",
        "image": "nginx:latest",
        "volumeMounts": [{
            "name": "config-volume",
            "mountPath": "/etc/config"
        }]
    }],
    volumes=[{
        "name": "config-volume",
        "configMap": {
            "name": "my-config"
        }
    }]
)
pod.create()

Pod with Node Selector

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

client = get_client()

pod = Pod(
    client=client,
    name="scheduled-pod",
    namespace="default",
    containers=[{
        "name": "nginx",
        "image": "nginx:latest"
    }],
    node_selector={
        "disktype": "ssd",
        "kubernetes.io/hostname": "node-1"
    }
)
pod.create()

Waiting for Pod to be Ready

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",
    containers=[{"name": "nginx", "image": "nginx:latest"}]
)
pod.create()

# Wait for pod to be running
pod.wait_for_status(status="Running", timeout=300)

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

print(f"Pod is running on node: {pod.node.name}")

Using Context Manager

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

client = get_client()

with Pod(
    client=client,
    name="temp-pod",
    namespace="default",
    containers=[{"name": "busybox", "image": "busybox:latest", "command": ["sleep", "3600"]}],
    wait_for_resource=True
) as pod:
    output = pod.execute(command=["echo", "Hello from pod"])
    print(output)
    # Pod is automatically deleted when exiting this block

See Also

• NamespacedResource - Base class
• Deployment - For managing pod replicas
• Service - For exposing pods
• ConfigMap - For configuration data
• Secret - For sensitive data