Skip to main content

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

Build docs developers (and LLMs) love