kopf.on module

The decorators for the event handlers. Usually used as:

import kopf

@kopf.on.create('kopfexamples')
def creation_handler(**kwargs):
    pass

This module is a part of the framework’s public interface.

kopf.on.startup(*, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, registry=None)[source]
Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.cleanup(*, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, registry=None)[source]
Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.login(*, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, registry=None)[source]

@kopf.on.login() handler for custom (re-)authentication.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.probe(*, id=None, param=None, errors=None, timeout=None, retries=None, backoff=None, registry=None)[source]

@kopf.on.probe() handler for arbitrary liveness metrics.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.validate(__group_or_groupversion_or_name=None, __version_or_name=None, __name=None, *, group=None, version=None, kind=None, plural=None, singular=None, shortcut=None, category=None, id=None, param=None, operation=None, subresource=None, persistent=None, side_effects=None, ignore_failures=None, labels=None, annotations=None, when=None, field=None, value=None, registry=None)[source]

@kopf.on.validate() handler for validating admission webhooks.

Return type

Callable[[Callable[..., Optional[Coroutine[None, None, None]]]], Callable[..., Optional[Coroutine[None, None, None]]]]

Parameters
kopf.on.mutate(__group_or_groupversion_or_name=None, __version_or_name=None, __name=None, *, group=None, version=None, kind=None, plural=None, singular=None, shortcut=None, category=None, id=None, param=None, operation=None, subresource=None, persistent=None, side_effects=None, ignore_failures=None, labels=None, annotations=None, when=None, field=None, value=None, registry=None)[source]

@kopf.on.mutate() handler for mutating admission webhooks.

Return type

Callable[[Callable[..., Optional[Coroutine[None, None, None]]]], Callable[..., Optional[Coroutine[None, None, None]]]]

Parameters
kopf.on.resume(__group_or_groupversion_or_name=None, __version_or_name=None, __name=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, deleted=None, labels=None, annotations=None, when=None, field=None, value=None, registry=None)[source]

@kopf.on.resume() handler for the object resuming on operator (re)start.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.create(__group_or_groupversion_or_name=None, __version_or_name=None, __name=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.on.create() handler for the object creation.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.update(__group_or_groupversion_or_name=None, __version_or_name=None, __name=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, old=None, new=None, registry=None)[source]

@kopf.on.update() handler for the object update or change.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.delete(__group_or_groupversion_or_name=None, __version_or_name=None, __name=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, optional=None, labels=None, annotations=None, when=None, field=None, value=None, registry=None)[source]

@kopf.on.delete() handler for the object deletion.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.field(__group_or_groupversion_or_name=None, __version_or_name=None, __name=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, value=None, old=None, new=None, registry=None)[source]

@kopf.on.field() handler for the individual field changes.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.index(__group_or_groupversion_or_name=None, __version_or_name=None, __name=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[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.event(__group_or_groupversion_or_name=None, __version_or_name=None, __name=None, *, group=None, version=None, kind=None, plural=None, singular=None, shortcut=None, category=None, id=None, param=None, labels=None, annotations=None, when=None, field=None, value=None, registry=None)[source]

@kopf.on.event() handler for the silent spies on the events.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.daemon(__group_or_groupversion_or_name=None, __version_or_name=None, __name=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[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.timer(__group_or_groupversion_or_name=None, __version_or_name=None, __name=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[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.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 efficiently a syntax sugar to look like all other handlers:

@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-cycle.

Return type

Callable[[Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]], Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]]

Parameters
kopf.on.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:

@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 efficiently an equivalent for:

@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

Callable[..., Union[object, None, Coroutine[None, None, Optional[object]]]]

Parameters