pykube package

Submodules

pykube.config module

Configuration code.

class pykube.config.BytesOrFile(filename=None, data=None, kubeconfig_path=None)

Bases: object

Implements the same interface for files and byte input.

bytes()

Returns the provided data as bytes.

filename()

Returns the provided data as a file location.

classmethod maybe_set(d, key, kubeconfig_path)

class pykube.config.KubeConfig(doc, current_context=None)

Bases: object

Main configuration class.

cluster

Returns the current selected cluster by exposing as a read-only property.

clusters

Returns known clusters by exposing as a read-only property.

contexts

Returns known contexts by exposing as a read-only property.

current_context

filepath = None

classmethod from_env()

Convenience function to create an instance of KubeConfig from the current environment.

First tries to use in-cluster ServiceAccount, then tries default ~/.kube/config (or KUBECONFIG)

classmethod from_file(filename=None, **kwargs)

Creates an instance of the KubeConfig class from a kubeconfig file.

Parameters:filename – The full path to the configuration file. Defaults to ~/.kube/config

classmethod from_service_account(path='/var/run/secrets/kubernetes.io/serviceaccount', **kwargs)

Construct KubeConfig from in-cluster service account.

classmethod from_url(url, **kwargs)

Creates an instance of the KubeConfig class from a single URL (useful for interacting with kubectl proxy).

kubeconfig_file

Returns the path to kubeconfig file as string, if it exists

kubeconfig_path

Returns the path to kubeconfig file, if it exists

namespace

Returns the current context namespace by exposing as a read-only property.

persist_doc()

reload()

set_current_context(value)

Sets the context to the provided value.

Parameters:

• value: The value for the current context

user

Returns the current user set by current context

users

Returns known users by exposing as a read-only property.

pykube.console module

pykube.console.main(argv=None)

Run the interactive Pykube console (usually invoked via python3 -m pykube)

pykube.exceptions module

Exceptions.

exception pykube.exceptions.HTTPError(code, message)

Bases: pykube.exceptions.PyKubeError

exception pykube.exceptions.KubernetesError

Bases: Exception

Base exception for all Kubernetes errors.

exception pykube.exceptions.ObjectDoesNotExist

Bases: pykube.exceptions.PyKubeError

exception pykube.exceptions.PyKubeError

Bases: pykube.exceptions.KubernetesError

PyKube specific errors.

pykube.http module

HTTP request related code.

class pykube.http.HTTPClient(config: pykube.config.KubeConfig, timeout: float = 10, dry_run: bool = False, verify: bool = True, http_adapter: Optional[requests.adapters.HTTPAdapter] = None)

Bases: object

Client for interfacing with the Kubernetes API.

delete(*args, **kwargs)

Executes an HTTP DELETE.

Parameters:

• args: Non-keyword arguments
• kwargs: Keyword arguments

get(*args, **kwargs)

Executes an HTTP GET.

Parameters:

• args: Non-keyword arguments
• kwargs: Keyword arguments

get_kwargs(**kwargs) → dict

Creates a full URL to request based on arguments.

Parametes:

• kwargs: All keyword arguments to build a kubernetes API endpoint

head(*args, **kwargs)

Executes an HTTP HEAD.

Parameters:

• args: Non-keyword arguments
• kwargs: Keyword arguments

http_adapter_cls

alias of KubernetesHTTPAdapter

options(*args, **kwargs)

Executes an HTTP OPTIONS.

Parameters:

• args: Non-keyword arguments
• kwargs: Keyword arguments

patch(*args, **kwargs)

Executes an HTTP PATCH.

Parameters:

• args: Non-keyword arguments
• kwargs: Keyword arguments

post(*args, **kwargs)

Executes an HTTP POST.

Parameters:

• args: Non-keyword arguments
• kwargs: Keyword arguments

put(*args, **kwargs)

Executes an HTTP PUT.

Parameters:

• args: Non-keyword arguments
• kwargs: Keyword arguments

raise_for_status(resp)

request(*args, **kwargs)

Makes an API request based on arguments.

Parameters:

• args: Non-keyword arguments
• kwargs: Keyword arguments

resource_list(api_version)

url

version

Get Kubernetes API version

class pykube.http.KubernetesHTTPAdapter(kube_config: pykube.config.KubeConfig, **kwargs)

Bases: requests.adapters.HTTPAdapter

send(request, **kwargs)

Sends PreparedRequest object. Returns Response object.

Parameters:

• request – The PreparedRequest being sent.
• stream – (optional) Whether to stream the request content.
• timeout (float or tuple or urllib3 Timeout object) – (optional) How long to wait for the server to send data before giving up, as a float, or a (connect timeout, read timeout) tuple.
• verify – (optional) Either a boolean, in which case it controls whether we verify the server’s TLS certificate, or a string, in which case it must be a path to a CA bundle to use
• cert – (optional) Any user-provided SSL certificate to be trusted.
• proxies – (optional) The proxies dictionary to apply to the request.

Return type:

requests.Response

pykube.mixins module

class pykube.mixins.ReplicatedMixin

Bases: object

replicas

scalable_attr = 'replicas'

class pykube.mixins.ScalableMixin

Bases: object

scalable

scale(replicas=None)

pykube.objects module

class pykube.objects.APIObject(api: pykube.http.HTTPClient, obj: dict)

Bases: object

Baseclass for all Kubernetes API objects

annotations

Annotations of the Kubernetes resource (metadata.annotations)

Annotations is an unstructured key value map stored with a resource that may be set by external tools to store and retrieve arbitrary metadata. They are not queryable and should be preserved when modifying objects. More info: http://kubernetes.io/docs/user-guide/annotations

api_kwargs(**kwargs)

base = None

create()

delete(propagation_policy: str = None)

Delete the Kubernetes resource by calling the API.

The parameter propagation_policy defines whether to cascade the delete. It can be “Foreground”, “Background” or “Orphan”. See https://kubernetes.io/docs/concepts/workloads/controllers/garbage-collection/#setting-the-cascading-deletion-policy

exists(ensure=False)

labels

Labels of the Kubernetes resource (metadata.labels)

Map of string keys and values that can be used to organize and categorize (scope and select) objects. May match selectors of replication controllers and services. More info: http://kubernetes.io/docs/user-guide/labels

metadata

name

Name of the Kubernetes resource (metadata.name)

Name must be unique within a namespace. Is required when creating resources, although some resources may allow a client to request the generation of an appropriate name automatically. Name is primarily intended for creation idempotence and configuration definition. Cannot be updated. More info: http://kubernetes.io/docs/user-guide/identifiers#names

namespace

objects

patch(strategic_merge_patch, *, subresource=None)

Patch the Kubernetes resource by calling the API with a “strategic merge” patch.

reload()

set_obj(obj: dict)

update(is_strategic=True, *, subresource=None)

Update the Kubernetes resource by calling the API (patch)

watch()

class pykube.objects.ClusterRole(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.APIObject

endpoint = 'clusterroles'

kind = 'ClusterRole'

version = 'rbac.authorization.k8s.io/v1'

class pykube.objects.ClusterRoleBinding(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.APIObject

endpoint = 'clusterrolebindings'

kind = 'ClusterRoleBinding'

version = 'rbac.authorization.k8s.io/v1'

class pykube.objects.ConfigMap(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'configmaps'

kind = 'ConfigMap'

version = 'v1'

class pykube.objects.CronJob(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'cronjobs'

kind = 'CronJob'

version = 'batch/v1beta1'

class pykube.objects.CustomResourceDefinition(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.APIObject

endpoint = 'customresourcedefinitions'

kind = 'CustomResourceDefinition'

version = 'apiextensions.k8s.io/v1'

class pykube.objects.DaemonSet(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'daemonsets'

kind = 'DaemonSet'

version = 'apps/v1'

class pykube.objects.Deployment(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject, pykube.mixins.ReplicatedMixin, pykube.mixins.ScalableMixin

endpoint = 'deployments'

kind = 'Deployment'

ready

rollout_undo(target_revision=None)

Produces same action as kubectl rollout undo deployment command. Input variable is revision to rollback to (in kubectl, –to-revision)

version = 'apps/v1'

class pykube.objects.Endpoint(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'endpoints'

kind = 'Endpoint'

version = 'v1'

class pykube.objects.Event(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'events'

kind = 'Event'

version = 'v1'

class pykube.objects.HorizontalPodAutoscaler(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'horizontalpodautoscalers'

kind = 'HorizontalPodAutoscaler'

version = 'autoscaling/v1'

class pykube.objects.Ingress(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'ingresses'

kind = 'Ingress'

version = 'networking.k8s.io/v1beta1'

class pykube.objects.Job(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject, pykube.mixins.ScalableMixin

endpoint = 'jobs'

kind = 'Job'

parallelism

scalable_attr = 'parallelism'

version = 'batch/v1'

class pykube.objects.LimitRange(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'limitranges'

kind = 'LimitRange'

version = 'v1'

class pykube.objects.Namespace(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.APIObject

endpoint = 'namespaces'

kind = 'Namespace'

version = 'v1'

class pykube.objects.NamespacedAPIObject(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.APIObject

namespace

Namespace scope of the Kubernetes resource (metadata.namespace)

Namespace defines the space within each name must be unique. Cannot be updated. More info: http://kubernetes.io/docs/user-guide/namespaces

class pykube.objects.Node(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.APIObject

cordon()

endpoint = 'nodes'

kind = 'Node'

uncordon()

unschedulable

version = 'v1'

class pykube.objects.ObjectManager

Bases: object

class pykube.objects.PersistentVolume(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.APIObject

endpoint = 'persistentvolumes'

kind = 'PersistentVolume'

version = 'v1'

class pykube.objects.PersistentVolumeClaim(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'persistentvolumeclaims'

kind = 'PersistentVolumeClaim'

version = 'v1'

class pykube.objects.Pod(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'pods'

kind = 'Pod'

logs(container=None, pretty=None, previous=False, since_seconds=None, since_time=None, timestamps=False, tail_lines=None, limit_bytes=None)

Produces the same result as calling kubectl logs pod/<pod-name>. Check parameters meaning at http://kubernetes.io/docs/api-reference/v1/operations/, part ‘read log of the specified Pod’. The result is plain text.

ready

version = 'v1'

class pykube.objects.PodDisruptionBudget(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'poddisruptionbudgets'

kind = 'PodDisruptionBudget'

version = 'policy/v1beta1'

class pykube.objects.PodSecurityPolicy(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.APIObject

endpoint = 'podsecuritypolicies'

kind = 'PodSecurityPolicy'

version = 'policy/v1beta1'

class pykube.objects.ReplicaSet(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject, pykube.mixins.ReplicatedMixin, pykube.mixins.ScalableMixin

endpoint = 'replicasets'

kind = 'ReplicaSet'

version = 'apps/v1'

class pykube.objects.ReplicationController(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject, pykube.mixins.ReplicatedMixin, pykube.mixins.ScalableMixin

endpoint = 'replicationcontrollers'

kind = 'ReplicationController'

ready

version = 'v1'

class pykube.objects.ResourceQuota(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'resourcequotas'

kind = 'ResourceQuota'

version = 'v1'

class pykube.objects.Role(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'roles'

kind = 'Role'

version = 'rbac.authorization.k8s.io/v1'

class pykube.objects.RoleBinding(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'rolebindings'

kind = 'RoleBinding'

version = 'rbac.authorization.k8s.io/v1'

class pykube.objects.Secret(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'secrets'

kind = 'Secret'

version = 'v1'

class pykube.objects.Service(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'services'

kind = 'Service'

proxy_http_delete(path: str, port: Optional[int] = None, **kwargs) → requests.models.Response

Issue a HTTP DELETE request to proxy of a Service. Args: :rtype: Response :type path: str

param path:The URI path for the request. param port:This value can be used to override the

default (first defined) port used to connect to the Service. :param kwargs: Keyword arguments for the proxy_http_get function. They are the same as for requests.models.Request object plus the additional ‘port’ kwarg, which can be used to override the default (first defined) port used to connect to the Service.

Returns:

The requests.Response object.

proxy_http_get(path: str, port: Optional[int] = None, **kwargs) → requests.models.Response

Issue a HTTP GET request to proxy of a Service. Args: :rtype: Response :type path: str

param path:The URI path for the request. param port:This value can be used to override the

default (first defined) port used to connect to the Service. :param kwargs: Keyword arguments for the proxy_http_get function. They are the same as for requests.models.Request object plus the additional ‘port’ kwarg, which can be used to override the default (first defined) port used to connect to the Service.

Returns:

The requests.Response object.

proxy_http_post(path: str, port: Optional[int] = None, **kwargs) → requests.models.Response

Issue a HTTP POST request to proxy of a Service. Args: :rtype: Response :type path: str

param path:The URI path for the request. param port:This value can be used to override the

default (first defined) port used to connect to the Service. :param kwargs: Keyword arguments for the proxy_http_get function. They are the same as for requests.models.Request object plus the additional ‘port’ kwarg, which can be used to override the default (first defined) port used to connect to the Service.

Returns:

The requests.Response object.

proxy_http_put(path: str, port: Optional[int] = None, **kwargs) → requests.models.Response

Issue a HTTP PUT request to proxy of a Service. Args: :rtype: Response :type path: str

param path:The URI path for the request. param port:This value can be used to override the

default (first defined) port used to connect to the Service. :param kwargs: Keyword arguments for the proxy_http_get function. They are the same as for requests.models.Request object plus the additional ‘port’ kwarg, which can be used to override the default (first defined) port used to connect to the Service.

Returns:

The requests.Response object.

proxy_http_request(method: str, path: str, port: Optional[int] = None, **kwargs) → requests.models.Response

Issue a HTTP request with specific HTTP method to proxy of a Service. Args: :rtype: Response :type path: str :type method: str

param method:The http request method e.g. ‘GET’, ‘POST’ etc. param path:The URI path for the request. param port:This value can be used to override the

default (first defined) port used to connect to the Service. :param kwargs: Keyword arguments for the proxy_http_get function. They are the same as for requests.models.Request object.

Returns:

The requests.Response object.

version = 'v1'

class pykube.objects.ServiceAccount(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject

endpoint = 'serviceaccounts'

kind = 'ServiceAccount'

version = 'v1'

class pykube.objects.StatefulSet(api: pykube.http.HTTPClient, obj: dict)

Bases: pykube.objects.NamespacedAPIObject, pykube.mixins.ReplicatedMixin, pykube.mixins.ScalableMixin

endpoint = 'statefulsets'

kind = 'StatefulSet'

version = 'apps/v1'

pykube.objects.object_factory(api, api_version, kind) → Type[pykube.objects.APIObject]

Dynamically builds a Python class for the given Kubernetes object in an API.

For example:

api = pykube.HTTPClient(…) NetworkPolicy = pykube.object_factory(api, “networking.k8s.io/v1”, “NetworkPolicy”)

This enables construction of any Kubernetes object kind without explicit support from pykube.

Currently, the HTTPClient passed to this function will not be bound to the returned type. It is planned to fix this, but in the mean time pass it as you would normally.

pykube.query module

class pykube.query.BaseQuery(api: pykube.http.HTTPClient, api_obj_class, namespace: str = None)

Bases: object

all() → pykube.query.BaseQuery

filter(namespace: str = None, selector: Union[str, dict] = None, field_selector: Union[str, dict] = None) → pykube.query.BaseQuery

Filter objects by namespace, labels, or fields

Parameters:

• namespace (str) – Namespace to filter by (pass pykube.all to get objects in all namespaces)
• selector – Label selector, can be a dictionary of label names/values

class pykube.query.Query(api: pykube.http.HTTPClient, api_obj_class, namespace: str = None)

Bases: pykube.query.BaseQuery

as_table() → pykube.query.Table

Execute query and return result as Table (similar to what kubectl does) See https://kubernetes.io/docs/reference/using-api/api-concepts/#receiving-resources-as-tables

execute(**kwargs)

get(*args, **kwargs)

Get a single object by name, namespace, label, ..

get_by_name(name: str)

Get object by name, raises ObjectDoesNotExist if not found

get_or_none(*args, **kwargs)

Get object by name, return None if not found

iterator()

Execute the API request and return an iterator over the objects. This method does not use the query cache.

query_cache

response

watch(since=None, *, params=None)

class pykube.query.Table(api_obj_class, obj: dict)

Bases: object

Tabular resource representation See https://kubernetes.io/docs/reference/using-api/api-concepts/#receiving-resources-as-tables

columns

rows

class pykube.query.WatchQuery(*args, **kwargs)

Bases: pykube.query.BaseQuery

object_stream()

response

pykube.query.as_selector(value: Union[str, dict]) → str

pykube.utils module

pykube.utils.join_url_path(*components, join_empty: bool = False) → str

Join given URL path components and return absolute path starting with ‘/’.

pykube.utils.jsonpath_parse(template, obj)

pykube.utils.obj_check(obj_value, original_obj_value, is_strategic=True)

pykube.utils.obj_merge(obj, original_obj, is_strategic=True)

Module contents

Python client for Kubernetes