tessl/pypi-kubernetes

Python client library for interacting with Kubernetes clusters through the Kubernetes API

—

Pending

Overview

Eval results

Files

Utilities

Name: tessl/pypi-kubernetes
Author: tessl

Helper functions for common Kubernetes operations including creating resources from YAML files, parsing quantity values, working with duration strings, and other utility functions that simplify common tasks when working with Kubernetes resources.

Capabilities

Resource Creation from Files

Create Kubernetes resources from YAML files, dictionaries, or directories containing multiple resource definitions.

def create_from_yaml(
    k8s_client,
    yaml_file: str,
    namespace: str = "default",
    verbose: bool = False,
    **kwargs
):
    """
    Create resources from YAML file.
    
    Parameters:
    - k8s_client: Kubernetes API client instance
    - yaml_file: Path to YAML file or file-like object
    - namespace: Default namespace for resources
    - verbose: Enable verbose output
    
    Returns:
    List of created resources
    """

def create_from_dict(
    k8s_client,
    data: dict,
    namespace: str = "default",
    verbose: bool = False,
    **kwargs
):
    """
    Create resources from Python dictionary.
    
    Parameters:
    - k8s_client: Kubernetes API client instance
    - data: Resource definition as dictionary
    - namespace: Default namespace for resources
    - verbose: Enable verbose output
    
    Returns:
    Created resource object
    """

def create_from_directory(
    k8s_client,
    yaml_dir: str,
    namespace: str = "default",
    verbose: bool = False,
    **kwargs
):
    """
    Create resources from directory of YAML files.
    
    Parameters:
    - k8s_client: Kubernetes API client instance
    - yaml_dir: Path to directory containing YAML files
    - namespace: Default namespace for resources
    - verbose: Enable verbose output
    
    Returns:
    List of created resources
    """

Quantity Parsing

Parse Kubernetes quantity strings for CPU and memory values into numeric formats.

def parse_quantity(quantity: str) -> float:
    """
    Parse Kubernetes quantity string to numeric value.
    
    Supports standard Kubernetes quantity formats:
    - CPU: 100m, 0.1, 1, 2.5
    - Memory: 128Mi, 1Gi, 512Ki, 1024
    - Storage: 10Gi, 1Ti, 500Mi
    
    Parameters:
    - quantity: Quantity string (e.g., "100m", "1Gi", "500Mi")
    
    Returns:
    Numeric value in base units (cores for CPU, bytes for memory/storage)
    """

Duration Parsing

Parse Kubernetes duration strings into seconds.

def parse_duration(duration: str) -> int:
    """
    Parse Kubernetes duration string to seconds.
    
    Supports formats: 1h30m45s, 90m, 3600s, etc.
    
    Parameters:
    - duration: Duration string (e.g., "1h30m", "45s", "2m30s")
    
    Returns:
    Duration in seconds
    """

Exception Types

class FailToCreateError(Exception):
    """
    Exception raised when resource creation fails.
    
    Attributes:
    - api_exceptions: List of API exceptions that occurred
    """
    api_exceptions: list

Usage Examples

Creating Resources from YAML

from kubernetes import client, config, utils

config.load_kube_config()
k8s_client = client.ApiClient()

# Create resources from single YAML file
try:
    utils.create_from_yaml(
        k8s_client, 
        "deployment.yaml",
        namespace="production"
    )
    print("Resources created successfully")
except utils.FailToCreateError as e:
    print(f"Failed to create resources: {e}")
    for api_exception in e.api_exceptions:
        print(f"  - {api_exception}")

# Example deployment.yaml content:
"""
apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx-deployment
spec:
  replicas: 3
  selector:
    matchLabels:
      app: nginx
  template:
    metadata:
      labels:
        app: nginx
    spec:
      containers:
      - name: nginx
        image: nginx:1.20
        ports:
        - containerPort: 80
---
apiVersion: v1
kind: Service
metadata:
  name: nginx-service
spec:
  selector:
    app: nginx
  ports:
  - port: 80
    targetPort: 80
  type: ClusterIP
"""

Creating Resources from Dictionary

from kubernetes import client, config, utils

config.load_kube_config()
k8s_client = client.ApiClient()

# Define resource as dictionary
pod_definition = {
    "apiVersion": "v1",
    "kind": "Pod",
    "metadata": {
        "name": "test-pod",
        "labels": {"app": "test"}
    },
    "spec": {
        "containers": [{
            "name": "test-container",
            "image": "nginx:latest",
            "resources": {
                "requests": {"cpu": "100m", "memory": "128Mi"},
                "limits": {"cpu": "500m", "memory": "512Mi"}
            }
        }]
    }
}

# Create resource
try:
    created_resource = utils.create_from_dict(
        k8s_client,
        pod_definition,
        namespace="default",
        verbose=True
    )
    print(f"Created pod: {created_resource.metadata.name}")
except utils.FailToCreateError as e:
    print(f"Failed to create pod: {e}")

Creating Resources from Directory

from kubernetes import client, config, utils
import os

config.load_kube_config()
k8s_client = client.ApiClient()

# Directory structure:
# manifests/
#   ├── deployment.yaml
#   ├── service.yaml
#   ├── configmap.yaml
#   └── ingress.yaml

try:
    created_resources = utils.create_from_directory(
        k8s_client,
        "manifests/",
        namespace="production",
        verbose=True
    )
    
    print(f"Created {len(created_resources)} resources:")
    for resource in created_resources:
        print(f"  - {resource.kind}: {resource.metadata.name}")
        
except utils.FailToCreateError as e:
    print(f"Failed to create some resources: {e}")
    print("Successfully created resources before failure:")
    for api_exception in e.api_exceptions:
        print(f"  Error: {api_exception}")

Parsing Resource Quantities

from kubernetes import utils

# Parse CPU quantities
cpu_millicores = utils.parse_quantity("500m")  # 0.5 cores
print(f"500m = {cpu_millicores} cores")

cpu_cores = utils.parse_quantity("2.5")  # 2.5 cores
print(f"2.5 cores = {cpu_cores} cores")

# Parse memory quantities
memory_bytes = utils.parse_quantity("512Mi")  # 536,870,912 bytes
print(f"512Mi = {memory_bytes} bytes")

memory_gi = utils.parse_quantity("2Gi")  # 2,147,483,648 bytes
print(f"2Gi = {memory_gi} bytes")

# Parse storage quantities
storage_bytes = utils.parse_quantity("10Gi")  # 10,737,418,240 bytes
print(f"10Gi = {storage_bytes} bytes")

# Convert back for display
def format_memory(bytes_value):
    """Convert bytes back to human readable format."""
    if bytes_value >= 1024**3:
        return f"{bytes_value / (1024**3):.1f}Gi"
    elif bytes_value >= 1024**2:
        return f"{bytes_value / (1024**2):.1f}Mi"
    elif bytes_value >= 1024:
        return f"{bytes_value / 1024:.1f}Ki"
    else:
        return f"{bytes_value}B"

print(f"Formatted: {format_memory(memory_bytes)}")

Parsing Duration Strings

from kubernetes import utils

# Parse different duration formats
duration_seconds = utils.parse_duration("1h30m45s")  # 5445 seconds
print(f"1h30m45s = {duration_seconds} seconds")

timeout_seconds = utils.parse_duration("5m")  # 300 seconds
print(f"5m = {timeout_seconds} seconds")

short_timeout = utils.parse_duration("30s")  # 30 seconds
print(f"30s = {short_timeout} seconds")

# Convert back to human readable
def format_duration(seconds):
    """Convert seconds back to human readable format."""
    hours = seconds // 3600
    minutes = (seconds % 3600) // 60
    secs = seconds % 60
    
    parts = []
    if hours > 0:
        parts.append(f"{hours}h")
    if minutes > 0:
        parts.append(f"{minutes}m")
    if secs > 0:
        parts.append(f"{secs}s")
    
    return "".join(parts) if parts else "0s"

print(f"Formatted: {format_duration(duration_seconds)}")

Resource Validation and Processing

from kubernetes import client, config, utils
import yaml

config.load_kube_config()
k8s_client = client.ApiClient()

def validate_and_create_resources(yaml_content):
    """Validate and create resources with error handling."""
    
    try:
        # Parse YAML content
        documents = yaml.safe_load_all(yaml_content)
        
        created_resources = []
        failed_resources = []
        
        for doc in documents:
            if not doc:  # Skip empty documents
                continue
                
            try:
                # Validate required fields
                if 'apiVersion' not in doc:
                    raise ValueError("Missing apiVersion")
                if 'kind' not in doc:
                    raise ValueError("Missing kind")
                if 'metadata' not in doc or 'name' not in doc['metadata']:
                    raise ValueError("Missing metadata.name")
                
                # Create resource
                created = utils.create_from_dict(
                    k8s_client,
                    doc,
                    namespace="default"
                )
                created_resources.append(created)
                print(f"✓ Created {doc['kind']}: {doc['metadata']['name']}")
                
            except Exception as e:
                failed_resources.append({
                    'resource': doc,
                    'error': str(e)
                })
                print(f"✗ Failed to create {doc.get('kind', 'Unknown')}: {e}")
        
        return created_resources, failed_resources
        
    except yaml.YAMLError as e:
        print(f"YAML parsing error: {e}")
        return [], []

# Example usage
yaml_content = """
apiVersion: v1
kind: ConfigMap
metadata:
  name: app-config
data:
  database_url: "postgresql://localhost:5432/mydb"
  debug: "true"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: app-deployment
spec:
  replicas: 2
  selector:
    matchLabels:
      app: myapp
  template:
    metadata:
      labels:
        app: myapp
    spec:
      containers:
      - name: app
        image: myapp:latest
        envFrom:
        - configMapRef:
            name: app-config
"""

created, failed = validate_and_create_resources(yaml_content)
print(f"\nSummary: {len(created)} created, {len(failed)} failed")

Working with Resource Templates

from kubernetes import client, config, utils
import yaml
from string import Template

config.load_kube_config()
k8s_client = client.ApiClient()

def create_from_template(template_file, values, namespace="default"):
    """Create resources from template file with variable substitution."""
    
    # Read template file
    with open(template_file, 'r') as f:
        template_content = f.read()
    
    # Substitute variables
    template = Template(template_content)
    rendered_content = template.substitute(values)
    
    # Parse and create resources
    documents = yaml.safe_load_all(rendered_content)
    created_resources = []
    
    for doc in documents:
        if doc:
            created = utils.create_from_dict(k8s_client, doc, namespace=namespace)
            created_resources.append(created)
    
    return created_resources

# Example template file (app-template.yaml):
"""
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ${app_name}-deployment
spec:
  replicas: ${replicas}
  selector:
    matchLabels:
      app: ${app_name}
  template:
    metadata:
      labels:
        app: ${app_name}
    spec:
      containers:
      - name: ${app_name}
        image: ${image}
        resources:
          requests:
            cpu: ${cpu_request}
            memory: ${memory_request}
"""

# Use template
template_values = {
    'app_name': 'my-web-app',
    'replicas': '3',
    'image': 'nginx:1.20',
    'cpu_request': '100m',
    'memory_request': '128Mi'
}

resources = create_from_template(
    'app-template.yaml',
    template_values,
    namespace='production'
)

print(f"Created {len(resources)} resources from template")

Install with Tessl CLI