kopf package

The main Kopf module for all the exported functions and classes.

kopf.subhandler(*, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, labels=None, annotations=None, when=None, field=None, value=None, old=None, new=None)[source]

@kopf.subhandler() decorator for the dynamically generated sub-handlers.

Can be used only inside of the handler function. It is effectively syntactic sugar to look like all other handlers:

import kopf

@kopf.on.create('kopfexamples')
def create(*, spec, **kwargs):

    for task in spec.get('tasks', []):

        @kopf.subhandler(id=f'task_{task}')
        def create_task(*, spec, task=task, **kwargs):
            pass

In this example, having spec.tasks set to [abc, def], this will create the following handlers: create, create/task_abc, create/task_def.

The parent handler is not considered as finished if there are unfinished sub-handlers left. Since the sub-handlers will be executed in the regular reactor and lifecycle, with multiple low-level events (one per iteration), the parent handler will also be executed multiple times, and is expected to produce the same (or at least predictable) set of sub-handlers. In addition, keep its logic idempotent (not failing on the repeated calls).

Note: task=task is needed to freeze the closure variable, so that every create function will have its own value, not the latest in the for-loop.

Return type:

Callable[[ChangingFn], ChangingFn]

Parameters:

  • id (str | None)

  • param (Any | None)

  • errors (ErrorsMode | None)

  • timeout (float | None)

  • retries (int | None)

  • backoff (float | None)

  • labels (Mapping[str, str | MetaFilterToken | MetaFilterFn] | None)

  • annotations (Mapping[str, str | MetaFilterToken | MetaFilterFn] | None)

  • when (WhenFilterFn | None)

  • field (str | tuple[str, ...] | list[str] | None)

  • value (Any | MetaFilterToken | MetaFilterFn | None)

  • old (Any | MetaFilterToken | MetaFilterFn | None)

  • new (Any | MetaFilterToken | MetaFilterFn | None)

kopf.register(fn, *, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, labels=None, annotations=None, when=None)[source]

Register a function as a sub-handler of the currently executed handler.

Example:

import kopf

@kopf.on.create('kopfexamples')
def create_it(spec, **kwargs):
    for task in spec.get('tasks', []):

        def create_single_task(task=task, **_):
            pass

        kopf.register(id=task, fn=create_single_task)

This is effectively equivalent to:

import kopf

@kopf.on.create('kopfexamples')
def create_it(spec, **kwargs):
    for task in spec.get('tasks', []):

        @kopf.subhandler(id=task)
        def create_single_task(task=task, **_):
            pass
Return type:

ChangingFn

Parameters:

  • fn (ChangingFn)

  • id (str | None)

  • param (Any | None)

  • errors (ErrorsMode | None)

  • timeout (float | None)

  • retries (int | None)

  • backoff (float | None)

  • labels (Mapping[str, str | MetaFilterToken | MetaFilterFn] | None)

  • annotations (Mapping[str, str | MetaFilterToken | MetaFilterFn] | None)

  • when (WhenFilterFn | None)

async kopf.execute(*, fns=None, handlers=None, registry=None, lifecycle=None, cause=None)[source]

Execute the handlers in an isolated lifecycle.

This function is a public entry point with multiple ways to specify the handlers: either as the raw functions, or as the pre-created handlers, or as a registry (as used in the object handling).

If no explicit functions or handlers or registry are passed, the sub-handlers of the current handler are assumed, as accumulated in the per-handler registry with @kopf.subhandler.

If the call to this method for the sub-handlers is not done explicitly in the handler, it is done implicitly after the handler is exited. One way or another, it is executed for the sub-handlers.

Return type:

None

Parameters:

  • fns (Iterable[ChangingFn] | None)

  • handlers (Iterable[ChangingHandler] | None)

  • registry (ChangingRegistry | None)

  • lifecycle (LifeCycleFn | None)

  • cause (Cause | None)

kopf.daemon(arg1=None, arg2=None, arg3=None, /, *, group=None, version=None, kind=None, plural=None, singular=None, shortcut=None, category=None, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, initial_delay=None, cancellation_backoff=None, cancellation_timeout=None, cancellation_polling=None, labels=None, annotations=None, when=None, field=None, value=None, registry=None)[source]

@kopf.daemon() decorator for the background threads/tasks.

Return type:

Callable[[DaemonFn], DaemonFn]

Parameters:

  • arg1 (str | Marker | Callable[[Resource], bool] | None)

  • arg2 (str | Marker | None)

  • arg3 (str | Marker | None)

  • group (str | None)

  • version (str | None)

  • kind (str | None)

  • plural (str | None)

  • singular (str | None)

  • shortcut (str | None)

  • category (str | None)

  • id (str | None)

  • param (Any | None)

  • errors (ErrorsMode | None)

  • timeout (float | None)

  • retries (int | None)

  • backoff (float | None)

  • initial_delay (float | DelayFn | None)

  • cancellation_backoff (float | None)

  • cancellation_timeout (float | None)

  • cancellation_polling (float | None)

  • labels (Mapping[str, str | MetaFilterToken | MetaFilterFn] | None)

  • annotations (Mapping[str, str | MetaFilterToken | MetaFilterFn] | None)

  • when (WhenFilterFn | None)

  • field (str | tuple[str, ...] | list[str] | None)

  • value (Any | MetaFilterToken | MetaFilterFn | None)

  • registry (OperatorRegistry | None)

kopf.timer(arg1=None, arg2=None, arg3=None, /, *, group=None, version=None, kind=None, plural=None, singular=None, shortcut=None, category=None, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, interval=None, initial_delay=None, sharp=None, idle=None, labels=None, annotations=None, when=None, field=None, value=None, registry=None)[source]

@kopf.timer() handler for the regular events.

Return type:

Callable[[TimerFn], TimerFn]

Parameters:
kopf.index(arg1=None, arg2=None, arg3=None, /, *, group=None, version=None, kind=None, plural=None, singular=None, shortcut=None, category=None, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, labels=None, annotations=None, when=None, field=None, value=None, registry=None)[source]

@kopf.index() handler for the indexing callbacks.

Return type:

Callable[[IndexingFn], IndexingFn]

Parameters:
kopf.configure(debug=None, verbose=None, quiet=None, log_format=LogFormat.FULL, log_prefix=False, log_refkey=None)[source]
Return type:

None

Parameters:
class kopf.LogFormat(*values)[source]

Bases: Enum

Log formats, as specified on CLI.

PLAIN = '%(message)s'[source]
FULL = '[%(asctime)s] %(name)-20.20s [%(levelname)-8.8s] %(message)s'[source]
JSON = '-json-'[source]
kopf.login_via_pykube(*, logger, settings, **_)[source]
Return type:

ConnectionInfo | None

Parameters:
kopf.login_via_client(*, logger, settings, **_)[source]
Return type:

ConnectionInfo | None

Parameters:
async kopf.login_via_async_client(*, logger, settings, **_)[source]
Return type:

ConnectionInfo | None

Parameters:
kopf.login_with_kubeconfig(*, settings, **_)[source]

A minimalistic login handler that can get raw data from a kubeconfig file.

Authentication capabilities can be limited to keep the code short & simple. No parsing or sophisticated multi-step token retrieval is performed.

This login function is intended to make Kopf runnable in trivial cases when neither pykube-ng nor the official client library are installed.

Return type:

ConnectionInfo | None

Parameters:
kopf.login_with_service_account(*, settings, **_)[source]

A minimalistic login handler that can get raw data from a service account.

Authentication capabilities can be limited to keep the code short & simple. No parsing or sophisticated multi-step token retrieval is performed.

This login function is intended to make Kopf runnable in trivial cases when neither pykube-ng nor the official client library are installed.

Return type:

ConnectionInfo | None

Parameters:
exception kopf.LoginError[source]

Bases: Exception

Raised when the operator cannot login to the API.

class kopf.ConnectionInfo(*, server, priority=0, default_namespace=None, expiration=None, ca_path=None, ca_data=None, insecure=None, username=None, password=None, scheme=None, token=None, certificate_path=None, certificate_data=None, private_key_path=None, private_key_data=None, proxy_url=None, trust_env=False)[source]

Bases: KubeContext

A single endpoint with specific credentials and connection flags to use.

Parameters:
server: str[source]
ca_path: str | bytes | PathLike[str] | PathLike[bytes] | None = None[source]
ca_data: str | bytes | None = None[source]
insecure: bool | None = None[source]
username: str | None = None[source]
password: str | None = None[source]
scheme: str | None = None[source]
token: str | None = None[source]
certificate_path: str | bytes | PathLike[str] | PathLike[bytes] | None = None[source]
certificate_data: str | bytes | None = None[source]
private_key_path: str | bytes | PathLike[str] | PathLike[bytes] | None = None[source]
private_key_data: str | bytes | None = None[source]
default_namespace: str | None = None[source]
proxy_url: str | None = None[source]
trust_env: bool = False[source]
priority: int = 0[source]
expiration: datetime | None = None[source]
as_aiohttp_basic_auth()[source]

Make a basic auth for username/password, or None if absent.

Return type:

BasicAuth | None

as_http_headers()[source]

Make a dict with the Authorization header set to scheme+token, or an empty dict if there are no tokens or schemes.

Return type:

dict[str, str]

as_ssl_context()[source]

Make an SSL context with CA and client cert using Python’s ssl.

Warning

It will store the kopf.ConnectionInfo.certificate_data and kopf.ConnectionInfo.private_key_data into temporary files for a brief moment of time until the SSL context is constructed, since Python’s ssl cannot load them from memory.

Return type:

SSLContext

class kopf.AiohttpSession(*, server, priority=0, default_namespace=None, expiration=None, aiohttp_session)[source]

Bases: KubeContext

A custom aiohttp session to use instead of the built-in one.

See: Custom HTTP sessions for details.

Parameters:

  • server (str)

  • priority (int)

  • default_namespace (str | None)

  • expiration (datetime | None)

  • aiohttp_session (ClientSession)

aiohttp_session: ClientSession[source]
kopf.event(objs, *, type, reason, message='')[source]
Return type:

None

Parameters:
kopf.info(objs, *, reason, message='')[source]
Return type:

None

Parameters:
kopf.warn(objs, *, reason, message='')[source]
Return type:

None

Parameters:
kopf.exception(objs, *, reason='', message='', exc=None)[source]
Return type:

None

Parameters:
async kopf.spawn_tasks(*, lifecycle=None, indexers=None, registry=None, settings=None, memories=None, insights=None, identity=None, standalone=None, priority=None, peering_name=None, liveness_endpoint=None, clusterwide=False, namespaces=(), namespace=None, stop_flag=None, ready_flag=None, vault=None, memo=None, _command=None)[source]

Spawn all the tasks needed to run the operator.

The tasks are properly inter-connected with the synchronisation primitives.

Return type:

Collection[Task]

Parameters:

  • lifecycle (LifeCycleFn | None)

  • indexers (OperatorIndexers | None)

  • registry (OperatorRegistry | None)

  • settings (OperatorSettings | None)

  • memories (ResourceMemories | None)

  • insights (Insights | None)

  • identity (Identity | None)

  • standalone (bool | None)

  • priority (int | None)

  • peering_name (str | None)

  • liveness_endpoint (str | None)

  • clusterwide (bool)

  • namespaces (Collection[str | Pattern[str]])

  • namespace (str | Pattern[str] | None)

  • stop_flag (Future | Event | Future[Any] | Event | None)

  • ready_flag (Future | Event | Future[Any] | Event | None)

  • vault (Vault | None)

  • memo (object | None)

  • _command (Coroutine[None, None, None] | None)

async kopf.run_tasks(root_tasks, *, ignored=frozenset({}))[source]

Orchestrate the tasks and terminate them gracefully when needed.

The root tasks are expected to run forever. Their number is limited. Once any of them exits, the whole operator and all other root tasks should exit.

The root tasks, in turn, can spawn multiple sub-tasks of various purposes. They can be awaited, monitored, or fired-and-forgot.

The hung tasks are those that were spawned during the operator runtime, and were not cancelled/exited on the root tasks termination. They are given some extra time to finish, after which they are forcibly terminated too.

Note

Due to implementation details, every task created after the operator’s startup is assumed to be a task or a sub-task of the operator. In the end, all tasks are forcibly cancelled. Even if those tasks were created by other means. There is no way to trace who spawned what. Only the tasks that existed before the operator startup are ignored (for example, those that spawned the operator itself).

Return type:

None

Parameters:
async kopf.operator(*, lifecycle=None, indexers=None, registry=None, settings=None, memories=None, insights=None, identity=None, standalone=None, priority=None, peering_name=None, liveness_endpoint=None, clusterwide=False, namespaces=(), namespace=None, stop_flag=None, ready_flag=None, vault=None, memo=None, _command=None)[source]

Run the whole operator asynchronously.

This function should be used to run an operator in an asyncio event-loop if the operator is orchestrated explicitly and manually.

It is effectively spawn_tasks() + run_tasks() with some safety.

Return type:

None

Parameters:

  • lifecycle (LifeCycleFn | None)

  • indexers (OperatorIndexers | None)

  • registry (OperatorRegistry | None)

  • settings (OperatorSettings | None)

  • memories (ResourceMemories | None)

  • insights (Insights | None)

  • identity (Identity | None)

  • standalone (bool | None)

  • priority (int | None)

  • peering_name (str | None)

  • liveness_endpoint (str | None)

  • clusterwide (bool)

  • namespaces (Collection[str | Pattern[str]])

  • namespace (str | Pattern[str] | None)

  • stop_flag (Future | Event | Future[Any] | Event | None)

  • ready_flag (Future | Event | Future[Any] | Event | None)

  • vault (Vault | None)

  • memo (object | None)

  • _command (Coroutine[None, None, None] | None)

kopf.run(*, loop=None, lifecycle=None, indexers=None, registry=None, settings=None, memories=None, insights=None, identity=None, standalone=None, priority=None, peering_name=None, liveness_endpoint=None, clusterwide=False, namespaces=(), namespace=None, stop_flag=None, ready_flag=None, vault=None, memo=None, _command=None)[source]

Run the whole operator synchronously.

If the loop is not specified, the operator runs in the event loop of the current _context_ (by asyncio’s default, the current thread). See: https://docs.python.org/3/library/asyncio-policy.html for details.

Alternatively, use asyncio.run(kopf.operator(...)) with the same args. It will take care of a new event loop’s creation and finalization for this call. See: asyncio.run().

Return type:

None

Parameters:

  • loop (AbstractEventLoop | None)

  • lifecycle (LifeCycleFn | None)

  • indexers (OperatorIndexers | None)

  • registry (OperatorRegistry | None)

  • settings (OperatorSettings | None)

  • memories (ResourceMemories | None)

  • insights (Insights | None)

  • identity (Identity | None)

  • standalone (bool | None)

  • priority (int | None)

  • peering_name (str | None)

  • liveness_endpoint (str | None)

  • clusterwide (bool)

  • namespaces (Collection[str | Pattern[str]])

  • namespace (str | Pattern[str] | None)

  • stop_flag (Future | Event | Future[Any] | Event | None)

  • ready_flag (Future | Event | Future[Any] | Event | None)

  • vault (Vault | None)

  • memo (object | None)

  • _command (Coroutine[None, None, None] | None)

kopf.adopt(objs, owner=None, *, forced=False, strict=False, nested=None)[source]

The children should be in the same namespace, named after their parent, and owned by it.

Return type:

None

Parameters:
kopf.label(objs, labels=_UNSET.token, *, forced=False, nested=None, force=None)[source]

Apply the labels to the object(s).

Return type:

None

Parameters:
kopf.not_(fn)[source]
Return type:

TypeVar(_FnT, WhenFilterFn, MetaFilterFn)

Parameters:

fn (_FnT)

kopf.all_(fns)[source]
Return type:

TypeVar(_FnT, WhenFilterFn, MetaFilterFn)

Parameters:

fns (Collection[_FnT])

kopf.any_(fns)[source]
Return type:

TypeVar(_FnT, WhenFilterFn, MetaFilterFn)

Parameters:

fns (Collection[_FnT])

kopf.none_(fns)[source]
Return type:

TypeVar(_FnT, WhenFilterFn, MetaFilterFn)

Parameters:

fns (Collection[_FnT])

kopf.get_default_lifecycle()[source]
Return type:

LifeCycleFn

kopf.set_default_lifecycle(lifecycle)[source]
Return type:

None

Parameters:

lifecycle (LifeCycleFn | None)

kopf.build_object_reference(body)[source]

Construct an object reference for the events.

Keep in mind that some fields can be absent: e.g. namespace for cluster resources, or e.g. apiVersion for kind: Node, etc.

Return type:

ObjectReference

Parameters:

body (Body)

kopf.build_owner_reference(body, *, controller=True, block_owner_deletion=True)[source]

Construct an owner reference object for the parent-children relationships.

The structure needed to link the children objects to the current object as a parent. See https://kubernetes.io/docs/concepts/workloads/controllers/garbage-collection/

Keep in mind that some fields can be absent: e.g. namespace for cluster resources, or e.g. apiVersion for kind: Node, etc.

Return type:

OwnerReference

Parameters:

  • body (Body)

  • controller (bool | None)

  • block_owner_deletion (bool | None)

kopf.append_owner_reference(objs, owner=None, *, controller=True, block_owner_deletion=True)[source]

Append an owner reference to the resource(s), if it is not yet there.

Note: the owned objects are usually not the one being processed, so the whole body can be modified, no patches are needed.

Return type:

None

Parameters:
kopf.remove_owner_reference(objs, owner=None)[source]

Remove an owner reference from the resource(s), if it is there.

Note: the owned objects are usually not the one being processed, so the whole body can be modified, no patches are needed.

Return type:

None

Parameters:
class kopf.ErrorsMode(*values)[source]

Bases: Enum

How arbitrary (non-temporary/non-permanent) exceptions are treated.

IGNORED = 1[source]
TEMPORARY = 2[source]
PERMANENT = 3[source]
exception kopf.AdmissionError(message='', code=500)[source]

Bases: PermanentError

Raised by admission handlers when an API operation under check is bad.

An admission error behaves the same as kopf.PermanentError, but provides admission-specific payload for the response: a message and a numeric code.

This error type is preferred when selecting only one error to report back to apiservers as the admission review result — in case multiple handlers are called in one admission request, i.e. when the webhook endpoints are not mapped to the handler ids (e.g. when configured manually).

Parameters:

  • message (str | None)

  • code (int | None)

Return type:

None

class kopf.WebhookClientConfigService[source]

Bases: TypedDict

namespace: str | None[source]
name: str | None[source]
path: str | None[source]
port: int | None[source]
class kopf.WebhookClientConfig[source]

Bases: TypedDict

A config of clients (apiservers) to access the webhooks’ server (operators).

This dictionary is put into managed webhook configurations “as is”. The fields & type annotations are only for hinting.

Kopf additionally modifies the url and the service’s path to inject handler ids as the last path component. This must be taken into account by custom webhook servers.

caBundle: str | None[source]
url: str | None[source]
service: WebhookClientConfigService | None[source]
class kopf.UserInfo[source]

Bases: TypedDict

username: str[source]
uid: str[source]
groups: list[str][source]
class kopf.WebhookFn(*args, **kwargs)[source]

Bases: Protocol

A framework-provided function to call when an admission request is received.

The framework provides the actual function. Custom webhook servers must accept the function, invoke it accordingly on admission requests, wait for the admission response, serialise it and send it back. They do not implement this function. This protocol only declares the exact signature.

class kopf.WebhookServer(*, addr=None, port=None, path=None, host=None, cadata=None, cafile=None, cadump=None, context=None, insecure=False, certfile=None, pkeyfile=None, password=None, extra_sans=(), verify_mode=None, verify_cafile=None, verify_capath=None, verify_cadata=None, file_check_interval=60.0)[source]

Bases: WebhookContextManager

A local HTTP/HTTPS endpoint.

Currently, the server is based on aiohttp, but the implementation can change in the future without warning.

This server is also used by specialised tunnels when they need a local endpoint to be tunneled.

  • addr, port is where to listen for connections (defaults to localhost and 9443).

  • path is the root path for a webhook server (defaults to no root path).

  • host is an optional override of the hostname for webhook URLs; if not specified, the addr will be used.

Kubernetes requires HTTPS, so HTTPS is the default mode of the server. This webhook server supports SSL both for the server certificates and for client certificates (e.g., for authentication) at the same time:

  • cadata, cafile is the CA bundle to be passed as a “client config” to the webhook configuration objects, to be used by clients/apiservers when talking to the webhook server; it is not used in the server itself.

  • cadump is a path to save the resulting CA bundle to be used by clients, i.e. apiservers; it can be passed to curl --cacert ...; if cafile is provided, it contains the same content.

  • certfile, pkeyfile define the server’s endpoint certificate; if not specified, a self-signed certificate and CA will be generated for both addr & host as SANs (but only host for CommonName).

  • password is either for decrypting the provided pkeyfile, or for encrypting and decrypting the generated private key.

  • extra_sans are put into the self-signed certificate as SANs (DNS/IP) in addition to the host & addr (in case some other endpoints exist).

  • verify_mode, verify_cafile, verify_capath, verify_cadata will be loaded into the SSL context for verifying the client certificates when provided and if provided by the clients, i.e. apiservers or curl; (ssl.SSLContext.verify_mode, ssl.SSLContext.load_verify_locations).

  • insecure flag disables HTTPS and runs an HTTP webhook server. This is used in ngrok for a local endpoint, but can be used for debugging or when the certificate-generating dependencies/extras are not installed.

Parameters:
DEFAULT_HOST: str | None = None[source]
addr: str | None[source]
port: int | None[source]
path: str | None[source]
host: str | None[source]
cadata: bytes | None[source]
cafile: str | PathLike | None[source]
cadump: str | PathLike | None[source]
context: SSLContext | None[source]
insecure: bool[source]
certfile: str | PathLike | None[source]
pkeyfile: str | PathLike | None[source]
password: str | None[source]
extra_sans: Iterable[str][source]
verify_mode: VerifyMode | None[source]
verify_cafile: str | PathLike | None[source]
verify_capath: str | PathLike | None[source]
verify_cadata: str | bytes | None[source]
file_check_interval: float[source]
static build_certificate(hostnames, password=None)[source]

Build a self-signed certificate with SANs (subject alternative names).

Returns a tuple of the certificate and its private key (PEM-formatted).

The certificate is “minimally sufficient”, without much of the extra information on the subject besides its common and alternative names. However, IP addresses are properly recognised and normalised for better compatibility with strict SSL clients (like apiservers of Kubernetes). The first non-IP hostname becomes the certificate’s common name – by convention, non-configurable. If no hostnames are found, the first IP address is used as a fallback. Magic IPs like 0.0.0.0 are excluded.

certbuilder is used as an implementation because it is lightweight: 2.9 MB vs. 8.7 MB for cryptography. Still, it is too heavy to include as a normal runtime dependency (for 8.8 MB of Kopf itself), so it is only available as the kopf[dev] extra for development-mode dependencies. This can change in the future if self-signed certificates become used at runtime (e.g. in production/staging environments or other real clusters).

Return type:

tuple[bytes, bytes]

Parameters:
class kopf.WebhookK3dServer(*, addr=None, port=None, path=None, host=None, cadata=None, cafile=None, cadump=None, context=None, insecure=False, certfile=None, pkeyfile=None, password=None, extra_sans=(), verify_mode=None, verify_cafile=None, verify_capath=None, verify_cadata=None, file_check_interval=60.0)[source]

Bases: WebhookServer

A tunnel from inside of K3d/K3s to its host where the operator is running.

With this tunnel, a developer can develop the webhooks when fully offline, since all the traffic is local and never leaves the host machine.

The forwarding is maintained by K3d itself. This tunnel only replaces the endpoints for the Kubernetes webhook and injects an SSL certificate with proper CN/SANs — to match Kubernetes’s SSL validity expectations.

Parameters:
DEFAULT_HOST: str | None = 'host.k3d.internal'[source]
class kopf.WebhookMinikubeServer(*, addr=None, port=None, path=None, host=None, cadata=None, cafile=None, cadump=None, context=None, insecure=False, certfile=None, pkeyfile=None, password=None, extra_sans=(), verify_mode=None, verify_cafile=None, verify_capath=None, verify_cadata=None, file_check_interval=60.0)[source]

Bases: WebhookServer

A tunnel from inside of Minikube to its host where the operator is running.

With this tunnel, a developer can develop the webhooks when fully offline, since all the traffic is local and never leaves the host machine.

The forwarding is maintained by Minikube itself. This tunnel only replaces the endpoints for the Kubernetes webhook and injects an SSL certificate with proper CN/SANs — to match Kubernetes’s SSL validity expectations.

Parameters:
DEFAULT_HOST: str | None = 'host.minikube.internal'[source]
class kopf.WebhookDockerDesktopServer(*, addr=None, port=None, path=None, host=None, cadata=None, cafile=None, cadump=None, context=None, insecure=False, certfile=None, pkeyfile=None, password=None, extra_sans=(), verify_mode=None, verify_cafile=None, verify_capath=None, verify_cadata=None, file_check_interval=60.0)[source]

Bases: WebhookServer

A tunnel from inside of Docker Desktop to its host where the operator is running.

With this tunnel, a developer can develop the webhooks when fully offline, since all the traffic is local and never leaves the host machine.

The forwarding is maintained by Docker Desktop itself. This tunnel only replaces the endpoints for the Kubernetes webhook and injects an SSL certificate with proper CN/SANs — to match Kubernetes’s SSL validity expectations.

Parameters:
DEFAULT_HOST: str | None = 'host.docker.internal'[source]
class kopf.WebhookNgrokTunnel(*, addr=None, port=None, path=None, token=None, region=None, binary=None)[source]

Bases: WebhookContextManager

Tunnel admission webhook requests via an external tunnel: ngrok.

addr, port, and path have the same meaning as in kopf.WebhookServer: where to listen for connections locally. Ngrok then tunnels this endpoint to a remote public URL.

Mind that the ngrok webhook tunnel runs the local webhook server in an insecure (HTTP) mode. For secure (HTTPS) mode, a paid subscription and properly issued certificates are needed. This goes beyond Kopf’s scope. If needed, implement your own ngrok tunnel.

Besides, ngrok tunnel does not report any CA to the webhook client configs. It is expected that the default trust chain is sufficient for ngrok’s certs.

token can be used for paid subscriptions, which lifts some limitations. Otherwise, the free plan has a limit of 40 requests per minute (this should be enough for local development).

binary, if set, will use the specified ngrok binary path; otherwise, pyngrok downloads the binary at runtime (not recommended).

Warning

The public URL is not properly protected and a malicious user can send requests to a locally running operator. If the handlers only process the data and make no side effects, this should be fine.

Despite ngrok providing basic auth (“username:password”), Kubernetes does not permit this information in the URLs.

Ngrok partially “protects” the URLS by assigning them random hostnames. Additionally, you can add random paths. However, this is not “security”, only a bit of safety for a short time (enough for development runs).

Parameters:
addr: str | None[source]
port: int | None[source]
path: str | None[source]
token: str | None[source]
region: str | None[source]
binary: str | PathLike[str] | None[source]
class kopf.WebhookAutoServer(*, addr=None, port=None, path=None, host=None, cadata=None, cafile=None, cadump=None, context=None, insecure=False, certfile=None, pkeyfile=None, password=None, extra_sans=(), verify_mode=None, verify_cafile=None, verify_capath=None, verify_cadata=None, file_check_interval=60.0)[source]

Bases: ClusterDetector, WebhookServer

A locally listening webserver which attempts to guess its proper hostname.

The choice is happening between supported webhook servers only (K3d/K3d and Minikube at the moment). In all other cases, a regular local server is started without hostname overrides.

If automatic tunneling is possible, consider WebhookAutoTunnel.

Parameters:
class kopf.WebhookAutoTunnel(*, addr=None, port=None, path=None)[source]

Bases: ClusterDetector, WebhookContextManager

The same as WebhookAutoServer, but with possible tunneling.

Generally, tunneling gives more possibilities to run in any environment, but it must not happen without a permission from the developers, and is not possible if running in a completely isolated/local/CI/CD cluster. Therefore, developers should activate automatic setup explicitly.

If automatic tunneling is prohibited or impossible, use WebhookAutoServer.

Note

Automatic server/tunnel detection is highly limited in configuration and provides only the most common options of all servers & tunnels: specifically, listening addr:port/path. All other options are specific to their servers/tunnels, and the auto-guessing logic cannot use/accept/pass them.

Parameters:

  • addr (str | None)

  • port (int | None)

  • path (str | None)

addr: str | None[source]
port: int | None[source]
path: str | None[source]
exception kopf.PermanentError[source]

Bases: Exception

A fatal handler error, the retries are useless.

exception kopf.TemporaryError(_TemporaryError__msg=None, delay=60)[source]

Bases: Exception

A potentially recoverable error, should be retried.

Parameters:

  • _TemporaryError__msg (str | None)

  • delay (float | None)

Return type:

None

exception kopf.HandlerTimeoutError[source]

Bases: PermanentError

An error for the handler’s timeout (if set).

exception kopf.HandlerRetriesError[source]

Bases: PermanentError

An error for the handler’s retries exceeded (if set).

class kopf.OperatorRegistry[source]

Bases: object

A global registry is used for handling of multiple resources & activities.

It is usually populated by the @kopf.on... decorators, but can also be explicitly created and used in the embedded operators.

kopf.get_default_registry()[source]

Get the default registry to be used by the decorators and the reactor unless the explicit registry is provided to them.

Return type:

OperatorRegistry

kopf.set_default_registry(registry)[source]

Set the default registry to be used by the decorators and the reactor unless the explicit registry is provided to them.

Return type:

None

Parameters:

registry (OperatorRegistry)

class kopf.OperatorSettings(process=<factory>, posting=<factory>, peering=<factory>, watching=<factory>, queueing=<factory>, scanning=<factory>, admission=<factory>, execution=<factory>, background=<factory>, networking=<factory>, persistence=<factory>)[source]

Bases: object

Parameters:

  • process (ProcessSettings)

  • posting (PostingSettings)

  • peering (PeeringSettings)

  • watching (WatchingSettings)

  • queueing (QueueingSettings)

  • scanning (ScanningSettings)

  • admission (AdmissionSettings)

  • execution (ExecutionSettings)

  • background (BackgroundSettings)

  • networking (NetworkingSettings)

  • persistence (PersistenceSettings)

process: ProcessSettings[source]
posting: PostingSettings[source]
peering: PeeringSettings[source]
watching: WatchingSettings[source]
queueing: QueueingSettings[source]
scanning: ScanningSettings[source]
admission: AdmissionSettings[source]
execution: ExecutionSettings[source]
background: BackgroundSettings[source]
networking: NetworkingSettings[source]
persistence: PersistenceSettings[source]
property batching: QueueingSettings[source]
class kopf.DiffBaseStorage(ignored_fields=None)[source]

Bases: StorageKeyMarkingConvention, StorageStanzaCleaner

Store the base essence for diff calculations, i.e. last handled state.

The “essence” is a snapshot of meaningful fields, which must be tracked to identify the actual changes on the object (or absence of such).

Used in the handling routines to check if there were significant changes (i.e. not the internal and system changes, like the uids, links, etc.), and to get the exact per-field diffs for the specific handler functions.

Conceptually similar to how kubectl apply stores the applied state on any object, and then uses that for the patch calculation: https://kubernetes.io/docs/concepts/overview/object-management-kubectl/declarative-config/

Parameters:

ignored_fields (Iterable[str | tuple[str, ...] | list[str] | None] | None)

build(*, body, extra_fields=None)[source]

Extract only the relevant fields for the state comparisons.

The framework ignores all the system fields (mostly from metadata) and the status stanza completely. Except for some well-known and useful metadata, such as labels and annotations (except for sure garbage).

A special set of fields can be provided even if they are supposed to be removed. This is used, for example, for handlers which react to changes in the specific fields in the status stanza, while the rest of the status stanza is removed.

It is generally not a good idea to override this method in custom stores, unless a different definition of an object’s essence is needed.

Return type:

BodyEssence

Parameters:
abstractmethod fetch(*, body)[source]
Return type:

BodyEssence | None

Parameters:

body (Body)

abstractmethod store(*, body, patch, essence)[source]
Return type:

None

Parameters:
class kopf.AnnotationsDiffBaseStorage(*, prefix='kopf.zalando.org', key='last-handled-configuration', ignored_fields=None, v1=True)[source]

Bases: StorageKeyFormingConvention, DiffBaseStorage

Parameters:
build(*, body, extra_fields=None)[source]

Extract only the relevant fields for the state comparisons.

The framework ignores all the system fields (mostly from metadata) and the status stanza completely. Except for some well-known and useful metadata, such as labels and annotations (except for sure garbage).

A special set of fields can be provided even if they are supposed to be removed. This is used, for example, for handlers which react to changes in the specific fields in the status stanza, while the rest of the status stanza is removed.

It is generally not a good idea to override this method in custom stores, unless a different definition of an object’s essence is needed.

Return type:

BodyEssence

Parameters:
fetch(*, body)[source]
Return type:

BodyEssence | None

Parameters:

body (Body)

store(*, body, patch, essence)[source]
Return type:

None

Parameters:
class kopf.StatusDiffBaseStorage(*, name='kopf', field='status.{name}.last-handled-configuration', ignored_fields=None)[source]

Bases: DiffBaseStorage

Parameters:
property field: tuple[str, ...][source]
build(*, body, extra_fields=None)[source]

Extract only the relevant fields for the state comparisons.

The framework ignores all the system fields (mostly from metadata) and the status stanza completely. Except for some well-known and useful metadata, such as labels and annotations (except for sure garbage).

A special set of fields can be provided even if they are supposed to be removed. This is used, for example, for handlers which react to changes in the specific fields in the status stanza, while the rest of the status stanza is removed.

It is generally not a good idea to override this method in custom stores, unless a different definition of an object’s essence is needed.

Return type:

BodyEssence

Parameters:
fetch(*, body)[source]
Return type:

BodyEssence | None

Parameters:

body (Body)

store(*, body, patch, essence)[source]
Return type:

None

Parameters:
class kopf.MultiDiffBaseStorage(storages)[source]

Bases: DiffBaseStorage

Parameters:

storages (Collection[DiffBaseStorage])

build(*, body, extra_fields=None)[source]

Extract only the relevant fields for the state comparisons.

The framework ignores all the system fields (mostly from metadata) and the status stanza completely. Except for some well-known and useful metadata, such as labels and annotations (except for sure garbage).

A special set of fields can be provided even if they are supposed to be removed. This is used, for example, for handlers which react to changes in the specific fields in the status stanza, while the rest of the status stanza is removed.

It is generally not a good idea to override this method in custom stores, unless a different definition of an object’s essence is needed.

Return type:

BodyEssence

Parameters:
fetch(*, body)[source]
Return type:

BodyEssence | None

Parameters:

body (Body)

store(*, body, patch, essence)[source]
Return type:

None

Parameters:
class kopf.ProgressRecord[source]

Bases: TypedDict

A single record stored for persistence of a single handler.

started: str | None[source]
stopped: str | None[source]
delayed: str | None[source]
purpose: str | None[source]
retries: int | None[source]
success: bool | None[source]
failure: bool | None[source]
message: str | None[source]
subrefs: Collection[HandlerId] | None[source]
class kopf.ProgressStorage[source]

Bases: StorageStanzaCleaner

Base class and an interface for all persistent states.

The state is persisted strictly per-handler, not for all handlers at once: to support overlapping operators (assuming different handler ids) storing their state on the same fields of the resource (e.g. state.kopf).

This also ensures that no extra logic for state merges will be needed: the handler states are atomic (i.e. state fields are not used separately) but independent: handlers should be persisted on their own, unrelated to other handlers, and never combined with other atomic structures.

If combining is still needed with performance optimization in mind (e.g. for relational/transactional databases), the keys can be cached in memory for short time, and flush() can be overridden to actually store them.

abstractmethod fetch(*, key, body)[source]
Return type:

ProgressRecord | None

Parameters:

  • key (HandlerId)

  • body (Body)

abstractmethod store(*, key, record, body, patch)[source]
Return type:

None

Parameters:
abstractmethod purge(*, key, body, patch)[source]
Return type:

None

Parameters:
abstractmethod touch(*, body, patch, value)[source]
Return type:

None

Parameters:
abstractmethod clear(*, essence)[source]
Return type:

BodyEssence

Parameters:

essence (BodyEssence)

flush()[source]
Return type:

None

class kopf.AnnotationsProgressStorage(*, prefix='kopf.zalando.org', verbose=False, touch_key='touch-dummy', v1=True)[source]

Bases: StorageKeyFormingConvention, StorageKeyMarkingConvention, ProgressStorage

State storage in .metadata.annotations with JSON-serialised content.

An example without a prefix:

An example with a prefix:

For the annotations’ naming conventions, hashing, and V1 & V2 differences, see AnnotationsNamingMixin.

Parameters:
fetch(*, key, body)[source]
Return type:

ProgressRecord | None

Parameters:

  • key (HandlerId)

  • body (Body)

store(*, key, record, body, patch)[source]
Return type:

None

Parameters:
purge(*, key, body, patch)[source]
Return type:

None

Parameters:
touch(*, body, patch, value)[source]
Return type:

None

Parameters:
clear(*, essence)[source]
Return type:

BodyEssence

Parameters:

essence (BodyEssence)

class kopf.StatusProgressStorage(*, name='kopf', field='status.{name}.progress', touch_field='status.{name}.dummy')[source]

Bases: ProgressStorage

State storage in .status stanza with deep structure.

The structure is this:

Parameters:
property field: tuple[str, ...][source]
property touch_field: tuple[str, ...][source]
fetch(*, key, body)[source]
Return type:

ProgressRecord | None

Parameters:

  • key (HandlerId)

  • body (Body)

store(*, key, record, body, patch)[source]
Return type:

None

Parameters:
purge(*, key, body, patch)[source]
Return type:

None

Parameters:
touch(*, body, patch, value)[source]
Return type:

None

Parameters:
clear(*, essence)[source]
Return type:

BodyEssence

Parameters:

essence (BodyEssence)

class kopf.MultiProgressStorage(storages)[source]

Bases: ProgressStorage

Parameters:

storages (Collection[ProgressStorage])

fetch(*, key, body)[source]
Return type:

ProgressRecord | None

Parameters:

  • key (HandlerId)

  • body (Body)

store(*, key, record, body, patch)[source]
Return type:

None

Parameters:
purge(*, key, body, patch)[source]
Return type:

None

Parameters:
touch(*, body, patch, value)[source]
Return type:

None

Parameters:
clear(*, essence)[source]
Return type:

BodyEssence

Parameters:

essence (BodyEssence)

class kopf.SmartProgressStorage(*, name='kopf', field='status.{name}.progress', touch_key='touch-dummy', touch_field='status.{name}.dummy', prefix='kopf.zalando.org', v1=True, verbose=False)[source]

Bases: MultiProgressStorage

Parameters:
class kopf.RawEvent[source]

Bases: TypedDict

type: Literal[None, 'ADDED', 'MODIFIED', 'DELETED', 'BOOKMARK'][source]
object: RawBody[source]
class kopf.RawBody[source]

Bases: TypedDict

apiVersion: str[source]
kind: str[source]
metadata: RawMeta[source]
spec: dict[str, Any][source]
status: dict[str, Any][source]
class kopf.Status(_Status__src)[source]

Bases: MappingView[str, Any]

Parameters:

_Status__src (Body)

class kopf.Spec(_Spec__src)[source]

Bases: MappingView[str, Any]

Parameters:

_Spec__src (Body)

class kopf.Meta(_Meta__src)[source]

Bases: MappingView[str, Any]

Parameters:

_Meta__src (Body)

property labels: Mapping[str, str][source]
property annotations: Mapping[str, str][source]
property uid: str[source]
property name: str[source]
property namespace: NamespaceName | None[source]
property creation_timestamp: str | None[source]
property deletion_timestamp: str | None[source]
class kopf.Body(_Body__src)[source]

Bases: ReplaceableMappingView[str, Any]

Parameters:

_Body__src (RawBody | BodyEssence)

property metadata: Meta[source]
property meta: Meta[source]
property spec: Spec[source]
property status: Status[source]
class kopf.BodyEssence[source]

Bases: TypedDict

metadata: MetaEssence[source]
spec: dict[str, Any][source]
status: dict[str, Any][source]
class kopf.ObjectReference[source]

Bases: TypedDict

apiVersion: str[source]
kind: str[source]
namespace: str | None[source]
name: str[source]
uid: str[source]
class kopf.OwnerReference[source]

Bases: TypedDict

controller: bool[source]
blockOwnerDeletion: bool[source]
apiVersion: str[source]
kind: str[source]
name: str[source]
uid: str[source]
class kopf.Memo[source]

Bases: dict[Any, Any]

A container to hold arbitrary keys-values assigned by operator developers.

It is used in the memo kwarg to all resource handlers, isolated per individual resource object (not the resource kind).

The values can be accessed either as dictionary keys (the memo is a dict under the hood) or as object attributes (except for methods of dict).

See more in In-memory containers.

>>> memo = Memo()

>>> memo.f1 = 100
>>> memo['f1']
... 100

>>> memo['f2'] = 200
>>> memo.f2
... 200

>>> set(memo.keys())
... {'f1', 'f2'}
class kopf.Index[source]

Bases: Mapping[_K, Store[_V]], Generic[_K, _V]

A mapping of index keys to collections of values indexed under those keys.

A single index is identified by a handler id and is populated by values usually from a single indexing function (the @kopf.index() decorator).

Note

This class is only an abstract interface of an index. The actual implementation is in .indexing.Index.

class kopf.Store[source]

Bases: Collection[_V], Generic[_V]

A collection of all values under a single unique index key.

Multiple objects can yield the same keys, so all their values are accumulated into collections. When an object is deleted or stops matching the filters, all associated values are discarded.

The order of values is not guaranteed.

The values are not deduplicated, so duplicates are possible if multiple objects return the same values from their indexing functions.

Note

This class is only an abstract interface of an indexed store. The actual implementation is in .indexing.Store.

class kopf.ObjectLogger(*, body, settings)[source]

Bases: LoggerAdapter

A logger/adapter to carry the object identifiers for formatting.

The identifiers are then used both for formatting the per-object messages in ObjectPrefixingFormatter, and when posting the k8s-events.

Constructed in event handling of each individual object.

The internal structure is made the same as an object reference in K8s API, but can change over time to anything needed for our internal purposes. However, as little information should be carried as possible, and the information should be protected against the object modification (e.g. in case of background posting via the queue; see K8sPoster).

Parameters:
process(msg, kwargs)[source]

Process the logging message and keyword arguments passed in to a logging call to insert contextual information. You can either manipulate the message itself, the keyword args or both. Return the message and kwargs modified (or not) to suit your needs.

Normally, you’ll only need to override this one method in a LoggerAdapter subclass for your specific needs.

Return type:

tuple[str, MutableMapping[str, Any]]

Parameters:
class kopf.LocalObjectLogger(*, body, settings)[source]

Bases: ObjectLogger

The same as ObjectLogger, but does not post the messages as k8s-events.

Used in the resource-watching handlers to log the handler’s invocation successes/failures without overloading K8s with excessively many k8s-events.

This class is used internally only and is not exposed publicly in any way.

Parameters:
log(*args, **kwargs)[source]

Delegate a log call to the underlying logger, after adding contextual information from this adapter instance.

Return type:

None

Parameters:
class kopf.Diff(_Diff__items)[source]

Bases: Sequence[DiffItem]

Parameters:

_Diff__items (Iterable[DiffItem])

class kopf.DiffItem(operation, field, old, new)[source]

Bases: NamedTuple

Parameters:
operation: DiffOperation[source]

Alias for field number 0

field: tuple[str, ...][source]

Alias for field number 1

old: Any[source]

Alias for field number 2

new: Any[source]

Alias for field number 3

property op: DiffOperation[source]
class kopf.DiffOperation(*values)[source]

Bases: str, Enum

ADD = 'add'[source]
CHANGE = 'change'[source]
REMOVE = 'remove'[source]
class kopf.Reason(*values)[source]

Bases: str, Enum

CREATE = 'create'[source]
UPDATE = 'update'[source]
DELETE = 'delete'[source]
RESUME = 'resume'[source]
NOOP = 'noop'[source]
FREE = 'free'[source]
GONE = 'gone'[source]
class kopf.Patch(src=None, /, body=None, fns=())[source]

Bases: dict[str, Any]

Parameters:
clear()[source]

Remove all items from the dict.

Return type:

None

property fns: list[Callable[[RawBody], None]][source]
property metadata: MetaPatch[source]
property meta: MetaPatch[source]
property spec: SpecPatch[source]
property status: StatusPatch[source]
as_json_patch(body=None)[source]

Build a list of JSON-patch ops for the changes & transformations.

As a reference resource body, either the argument is used (if provided), or the original resource body. But the reference body is mandatory — the patch calculates the differences relative to the reference body.

Some changes might disappear from the list if they are useless (no-op): e.g., setting a key to None to delete it when it is already absent; or setting the key to a value which is already in the resource body.

Return type:

list[JSONPatchItem]

Parameters:

body (Body | RawBody | None)

class kopf.DaemonStoppingReason(*values)[source]

Bases: Flag

A reason or combination of reasons for a daemon being terminated.

Daemons are signalled to exit usually for two reasons: the operator itself is exiting or restarting, so all daemons of all resources must stop; or the individual resource was deleted, but the operator continues running.

No matter the reason, the daemons must exit, so one and only one stop-flag is used. Some daemons can check the reason for exiting if it is important.

There can be multiple reasons combined (in rare cases, all of them).

DONE = 1[source]
FILTERS_MISMATCH = 2[source]
RESOURCE_DELETED = 4[source]
OPERATOR_PAUSING = 8[source]
OPERATOR_EXITING = 16[source]
DAEMON_SIGNALLED = 32[source]
DAEMON_CANCELLED = 64[source]
DAEMON_ABANDONED = 128[source]
class kopf.Resource(group, version, plural, kind=None, singular=None, shortcuts=frozenset({}), categories=frozenset({}), subresources=frozenset({}), namespaced=None, preferred=True, verbs=frozenset({}))[source]

Bases: object

A reference to a very specific custom or built-in resource kind.

It is used to form the K8s API URLs. Generally, K8s API only needs an API group, an API version, and a plural name of the resource. All other names are remembered to match against resource selectors, for logging, and for informational purposes.

Parameters:
group: str[source]

The resource’s API group; e.g. "kopf.dev", "apps", "batch". For Core v1 API resources, an empty string: "".

version: str[source]

The resource’s API version; e.g. "v1", "v1beta1", etc.

plural: str[source]

The resource’s plural name; e.g. "pods", "kopfexamples". It is used as an API endpoint, together with API group & version.

kind: str | None = None[source]

The resource’s kind (as in YAML files); e.g. "Pod", "KopfExample".

singular: str | None = None[source]

The resource’s singular name; e.g. "pod", "kopfexample".

shortcuts: frozenset[str] = frozenset({})[source]

The resource’s short names; e.g. {"po"}, {"kex", "kexes"}.

categories: frozenset[str] = frozenset({})[source]

The resource’s categories, to which the resource belongs; e.g. {"all"}.

subresources: frozenset[str] = frozenset({})[source]

The resource’s subresources, if defined; e.g. {"status", "scale"}.

namespaced: bool | None = None[source]

Whether the resource is namespaced (True) or cluster-scoped (False).

preferred: bool = True[source]

Whether the resource belong to a “preferred” API version. Only “preferred” resources are served when the version is not specified.

verbs: frozenset[str] = frozenset({})[source]

All available verbs for the resource, as supported by K8s API; e.g., {"list", "watch", "create", "update", "delete", "patch"}. Note that it is not the same as all verbs permitted by RBAC.

get_url(*, server=None, namespace=None, name=None, subresource=None, params=None)[source]

Build a URL to be used with K8s API.

If the namespace is not set, a cluster-wide URL is returned. For cluster-scoped resources, the namespace is ignored.

If the name is not set, the URL for the resource list is returned. Otherwise (if set), the URL for the individual resource is returned.

If subresource is set, that subresource’s URL is returned, regardless of whether such a subresource is known or not.

Params go to the query parameters (?param1=value1&param2=value2...).

Return type:

str

Parameters:

  • server (str | None)

  • namespace (NamespaceName | None)

  • name (str | None)

  • subresource (str | None)

  • params (dict[str, str] | None)

Submodules