How to use the kopf.reactor.registries function in kopf

def field(
        group: str, version: str, plural: str,
        field: Union[str, List[str], Tuple[str, ...]],
        *,
        id: Optional[str] = None,
        errors: Optional[registries.ErrorsMode] = None,
        timeout: Optional[float] = None,
        retries: Optional[int] = None,
        backoff: Optional[float] = None,
        cooldown: Optional[float] = None,  # deprecated, use `backoff`
        registry: Optional[registries.OperatorRegistry] = None,
        labels: Optional[bodies.Labels] = None,
        annotations: Optional[bodies.Annotations] = None,
) -> ResourceHandlerDecorator:
    """ ``@kopf.on.field()`` handler for the individual field changes. """
    actual_registry = registry if registry is not None else registries.get_default_registry()
    def decorator(fn: registries.ResourceHandlerFn) -> registries.ResourceHandlerFn:
        return actual_registry.register_resource_changing_handler(
            group=group, version=version, plural=plural,
            reason=None, field=field, id=id,
            errors=errors, timeout=timeout, retries=retries, backoff=backoff, cooldown=cooldown,
            fn=fn, labels=labels, annotations=annotations,
        )
    return decorator

cause: causation.ResourceChangingCause,
    ) -> Iterator[registries.ResourceHandler]:
        warnings.warn("SimpleRegistry.iter_cause_handlers() is deprecated; use "
                      "ResourceChangingRegistry.iter_handlers().", DeprecationWarning)

        changed_fields = frozenset(field for _, field, _, _ in cause.diff or [])
        for handler in self._handlers:
            if handler.reason is None or handler.reason == cause.reason:
                if handler.initial and not cause.initial:
                    pass  # ignore initial handlers in non-initial causes.
                elif registries.match(handler=handler, body=cause.body,
                                      changed_fields=changed_fields):
                    yield handler


class GlobalRegistry(BaseRegistry, registries.OperatorRegistry):
    """
    .. deprecated: 1.0

        Replaced with `MultiRegistry`.
    """

    def register_event_handler(self, *args: Any, **kwargs: Any) -> Any:
        warnings.warn("GlobalRegistry.register_event_handler() is deprecated; use "
                      "OperatorRegistry.register_resource_watching_handler().", DeprecationWarning)
        return self.register_resource_watching_handler(*args, **kwargs)

    def register_cause_handler(self, *args: Any, **kwargs: Any) -> Any:
        warnings.warn("GlobalRegistry.register_cause_handler() is deprecated; use "
                      "OperatorRegistry.register_resource_changing_handler().", DeprecationWarning)
        return self.register_resource_changing_handler(*args, **kwargs)

) -> Iterator[registries.ResourceHandler]:
        warnings.warn("GlobalRegistry.iter_event_handlers() is deprecated; use "
                      "OperatorRegistry.iter_resource_watching_handlers().", DeprecationWarning)
        cause = _create_watching_cause(resource=resource, event=event)
        yield from self.iter_resource_watching_handlers(cause=cause)

    def iter_cause_handlers(
            self,
            cause: causation.ResourceChangingCause,
    ) -> Iterator[registries.ResourceHandler]:
        warnings.warn("GlobalRegistry.iter_cause_handlers() is deprecated; use "
                      "OperatorRegistry.iter_resource_changing_handlers().", DeprecationWarning)
        yield from self.iter_resource_changing_handlers(cause=cause)


class SmartGlobalRegistry(registries.SmartOperatorRegistry, GlobalRegistry):
    pass


def _create_watching_cause(
        resource: resources_.Resource,
        event: bodies.Event,
) -> causation.ResourceWatchingCause:
    return causation.detect_resource_watching_cause(
        resource=resource,
        event=event,
        patch=patches.Patch(),  # unused
        type=event['type'],  # unused
        body=event['object'],  # unused
        raw=event,  # unused
    )

class HandlerTimeoutError(PermanentError):
    """ An error for the handler's timeout (if set). """


class HandlerRetriesError(PermanentError):
    """ An error for the handler's retries exceeded (if set). """


class HandlerChildrenRetry(TemporaryError):
    """ An internal pseudo-error to retry for the next sub-handlers attempt. """


# The task-local context; propagated down the stack instead of multiple kwargs.
# Used in `@kopf.on.this` and `kopf.execute()` to add/get the sub-handlers.
sublifecycle_var: ContextVar[lifecycles.LifeCycleFn] = ContextVar('sublifecycle_var')
subregistry_var: ContextVar[registries.ResourceChangingRegistry] = ContextVar('subregistry_var')
subexecuted_var: ContextVar[bool] = ContextVar('subexecuted_var')
handler_var: ContextVar[registries.BaseHandler] = ContextVar('handler_var')
cause_var: ContextVar[causation.BaseCause] = ContextVar('cause_var')


async def activity_trigger(
        *,
        lifecycle: lifecycles.LifeCycleFn,
        registry: registries.OperatorRegistry,
        activity: causation.Activity,
) -> Mapping[registries.HandlerId, registries.HandlerResult]:
    """
    Execute a handling cycle until succeeded or permanently failed.

    This mimics the behaviour of patching-watching in Kubernetes, but in-memory.
    """

async def _call_handler(
        handler: registries.BaseHandler,
        *args: Any,
        cause: causation.BaseCause,
        lifecycle: lifecycles.LifeCycleFn,
        **kwargs: Any,
) -> Optional[registries.HandlerResult]:
    """
    Invoke one handler only, according to the calling conventions.

    Specifically, calculate the handler-specific fields (e.g. field diffs).

    Ensure the global context for this asyncio task is set to the handler and
    its cause -- for proper population of the sub-handlers via the decorators
    (see `@kopf.on.this`).
    """

    # For the field-handlers, the old/new/diff values must match the field, not the whole object.
    if (True and  # for readable indenting
            isinstance(cause, causation.ResourceChangingCause) and
            isinstance(handler, registries.ResourceHandler) and
            handler.field is not None):
        old = dicts.resolve(cause.old, handler.field, None, assume_empty=True)

Handle a received event, log but ignore all errors.

    This is a lightweight version of the cause handling, but for the raw events,
    without any progress persistence. Multi-step calls are also not supported.
    If the handler fails, it fails and is never retried.

    Note: K8s-event posting is skipped for `kopf.on.event` handlers,
    as they should be silent. Still, the messages are logged normally.
    """
    handlers = registry.get_resource_watching_handlers(cause=cause)
    outcomes = await _execute_handlers(
        lifecycle=lifecycle,
        handlers=handlers,
        cause=cause,
        state=states.State.from_scratch(handlers=handlers),
        default_errors=registries.ErrorsMode.IGNORED,
    )

    # Store the results, but not the handlers' progress.
    states.deliver_results(outcomes=outcomes, patch=cause.patch)

async def _execute_handlers(
        lifecycle: lifecycles.LifeCycleFn,
        handlers: Collection[registries.BaseHandler],
        cause: causation.BaseCause,
        state: states.State,
        default_errors: registries.ErrorsMode = registries.ErrorsMode.TEMPORARY,
) -> Mapping[registries.HandlerId, states.HandlerOutcome]:
    """
    Call the next handler(s) from the chain of the handlers.

    Keep the record on the progression of the handlers in the object's state,
    and use it on the next invocation to determined which handler(s) to call.

    This routine is used both for the global handlers (via global registry),
    and for the sub-handlers (via a simple registry of the current handler).
    """

    # Filter and select the handlers to be executed right now, on this event reaction cycle.
    handlers_todo = [h for h in handlers if state[h.id].awakened]
    handlers_plan = await invocation.invoke(lifecycle, handlers_todo, cause=cause, state=state)

    # Execute all planned (selected) handlers in one event reaction cycle, even if there are few.
    outcomes: MutableMapping[registries.HandlerId, states.HandlerOutcome] = {}

import asyncio
import contextlib
import contextvars
import functools
from typing import Optional, Any, Union, List, Iterable, Iterator, Tuple

from kopf import config
from kopf.reactor import causation
from kopf.reactor import lifecycles
from kopf.reactor import registries
from kopf.structs import dicts

Invokable = Union[
    lifecycles.LifeCycleFn,
    registries.ActivityHandlerFn,
    registries.ResourceHandlerFn,
]


@contextlib.contextmanager
def context(
        values: Iterable[Tuple[contextvars.ContextVar[Any], Any]],
) -> Iterator[None]:
    """
    A context manager to set the context variables temporarily.
    """
    tokens: List[Tuple[contextvars.ContextVar[Any], contextvars.Token[Any]]] = []
    try:
        for var, val in values:
            token = var.set(val)
            tokens.append((var, token))
        yield

Login to Kubernetes cluster, locally or remotely.

    Keep the logged in state or config object in the global variable,
    so that it is available for future operator runs.
    """
    warnings.warn("kopf.login() is deprecated; the operator now authenticates "
                  "internally; cease using kopf.login().", DeprecationWarning)

    # Remember the credentials store for later usage in the actual operator.
    # Set the global vault for the legacy login()->run() scenario.
    global global_vault
    global_vault = credentials.Vault()

    # Perform the initial one-time authentication in presumably the same loop.
    loop = loop if loop is not None else asyncio.get_event_loop()
    registry = registries.get_default_registry()
    try:
        loop.run_until_complete(activities.authenticate(
            registry=registry,
            vault=global_vault,
        ))
    except asyncio.CancelledError:
        pass
    except handling.ActivityError as e:
        # Detect and re-raise the original LoginErrors, not the general activity error.
        # This is only needed for the legacy one-shot login, not for a background job.
        for outcome in e.outcomes.values():
            if isinstance(outcome.exception, credentials.LoginError):
                raise outcome.exception
        else:
            raise

def creation_handler(**kwargs):
        pass

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

# TODO: add cluster=True support (different API methods)

from typing import Optional, Callable, Union, Tuple, List

from kopf.reactor import causation
from kopf.reactor import handling
from kopf.reactor import registries
from kopf.structs import bodies

ResourceHandlerDecorator = Callable[[registries.ResourceHandlerFn], registries.ResourceHandlerFn]
ActivityHandlerDecorator = Callable[[registries.ActivityHandlerFn], registries.ActivityHandlerFn]


def startup(
        *,
        id: Optional[str] = None,
        errors: Optional[registries.ErrorsMode] = None,
        timeout: Optional[float] = None,
        retries: Optional[int] = None,
        backoff: Optional[float] = None,
        cooldown: Optional[float] = None,  # deprecated, use `backoff`
        registry: Optional[registries.OperatorRegistry] = None,
) -> ActivityHandlerDecorator:
    actual_registry = registry if registry is not None else registries.get_default_registry()
    def decorator(fn: registries.ActivityHandlerFn) -> registries.ActivityHandlerFn:
        return actual_registry.register_activity_handler(