In-memory containers¶

Kopf provides several ways of storing and exchanging the data in-memory between handlers and operators.

Resource memos¶

Every resource handler gets a memo kwarg of type kopf.Memo. It is an in-memory container for arbitrary runtime-only key-value pairs. The values can be accessed as either object attributes or dictionary keys.

The memo is shared by all handlers of the same individual resource (not of the resource kind, but a resource object). If the resource is deleted and re-created with the same name, the memo is also re-created (technically, it is a new resource).

import kopf
from typing import Any

@kopf.on.event('KopfExample')
def pinged(memo: kopf.Memo, **_: Any) -> None:
    memo.counter = memo.get('counter', 0) + 1

@kopf.timer('KopfExample', interval=10)
def tick(memo: kopf.Memo, logger: kopf.Logger, **_: Any) -> None:
    logger.info(f"{memo.counter} events have been received in 10 seconds.")
    memo.counter = 0

Operator memos¶

In operator handlers, such as startup/cleanup, liveness probes, credentials retrieval, and everything else not specific to resources, memo points to the operator’s global container for arbitrary values.

The per-operator container can be populated in the startup handlers, passed from outside the operator when Embedding is used, or both:

import kopf
import queue
import threading
from typing import Any

@kopf.on.startup()
def start_background_worker(memo: kopf.Memo, **_: Any) -> None:
    memo.my_queue = queue.Queue()
    memo.my_thread = threading.Thread(target=background, args=(memo.my_queue,))
    memo.my_thread.start()

@kopf.on.cleanup()
def stop_background_worker(memo: kopf.Memo, **_: Any) -> None:
    memo['my_queue'].put(None)
    memo['my_thread'].join()

def background(queue: queue.Queue) -> None:
    while True:
        item = queue.get()
        if item is None:
            break
        else:
            print(item)

Note

For code quality and style consistency, it is recommended to use the same approach when accessing the stored values. The mixed style here is for demonstration purposes only.

The operator’s memo is later used to populate per-resource memos. All keys and values are shallow-copied into each resource’s memo, where they can be mixed with per-resource values:

# ... continued from the previous example.
@kopf.on.event('KopfExample')
def pinged(memo: kopf.Memo, namespace: str | None, name: str, **_: Any) -> None:
    if not memo.get('is_seen'):
        memo.my_queue.put(f"{namespace}/{name}")
        memo.is_seen = True

Any changes to the operator’s container made after the first appearance of a resource are not replicated to existing resources’ containers, and are not guaranteed to be seen by new resources (even if they currently are).

However, due to shallow copying, mutable objects (lists, dicts, and even custom instances of kopf.Memo itself) in the operator’s container can be modified from outside, and these changes will be visible in all individual resource handlers and daemons that use their per-resource containers.

Custom memo classes¶

For embedded operators (Embedding), it is possible to use any class for memos. It is not even necessary to inherit from kopf.Memo.

There are 2 strict requirements:

The class must be supported by all involved handlers that use it.
The class must support shallow copying via copy.copy() (__copy__()).

The latter is used to create per-resource memos from the operator’s memo. To have one global memo shared by all individual resources, redefine the class to return self when asked to make a copy, as shown below:

import asyncio
import dataclasses
import kopf
from typing import Any

@dataclasses.dataclass()
class CustomContext:
    create_tpl: str
    delete_tpl: str

    def __copy__(self) -> "CustomContext":
        return self

@kopf.on.create('kopfexamples')
def create_fn(memo: CustomContext, **kwargs: Any) -> None:
    print(memo.create_tpl.format(**kwargs))

@kopf.on.delete('kopfexamples')
def delete_fn(memo: CustomContext, **kwargs: Any) -> None:
    print(memo.delete_tpl.format(**kwargs))

if __name__ == '__main__':
    kopf.configure(verbose=True)
    asyncio.run(kopf.operator(
        memo=CustomContext(
            create_tpl="Hello, {name}!",
            delete_tpl="Good bye, {name}!",
        ),
    ))

In all other respects, the framework does not use memos for its own needs and passes them through the call stack to the handlers and daemons as-is.

This advanced feature is not available for operators executed via kopf run.

Limitations¶

All in-memory values are lost on operator restarts; there is no persistence.

The in-memory containers are recommended only for ephemeral objects scoped to the process lifetime, such as concurrency primitives: locks, tasks, threads… For persistent values, use the status stanza or annotations of the resources.

Essentially, the operator’s memo is not much different from global variables (unless two or more embedded operator tasks are running) or asyncio contextvars, except that it provides the same interface as per-resource memos.