Replace registries with lists of handlers in the internal executor
@@ -19,7 +19,7 @@ import collections
import datetime
import logging
from contextvars import ContextVar
-from typing import Optional, Callable, Iterable, Union
+from typing import Optional, Callable, Iterable, Union, Collection
from kopf import events
from kopf.k8s import patching
@@ -185,8 +185,13 @@ async def handle_cause(
if cause.event in causation.HANDLER_CAUSES:
title = causation.TITLES.get(cause.event, repr(cause.event))
logger.debug(f"{title.capitalize()} event: %r", body)
+ handlers = registry.get_cause_handlers(cause=cause)
try:
- await execute(lifecycle=lifecycle, registry=registry, cause=cause)
+ await _execute(
+ lifecycle=lifecycle,
+ handlers=handlers,
+ cause=cause,
+ )
except HandlerChildrenRetry as e:
# on the top-level, no patches -- it is pre-patched.
delay = e.delay
@@ -295,14 +300,14 @@ async def execute(
# Raises `HandlerChildrenRetry` if the execute should be continued on the next iteration.
await _execute(
lifecycle=lifecycle,
- registry=registry,
+ handlers=registry.get_cause_handlers(cause=cause),
cause=cause,
)
async def _execute(
lifecycle: Callable,
- registry: registries.BaseRegistry,
+ handlers: Collection[registries.Handler],
cause: causation.Cause,
retry_on_errors: bool = True,
) -> None:
@@ -324,7 +329,6 @@ async def _execute(
logger = cause.logger
# Filter and select the handlers to be executed right now, on this event reaction cycle.
- handlers = registry.get_cause_handlers(cause=cause)
handlers_done = [h for h in handlers if status.is_finished(body=cause.body, handler=h)]
handlers_wait = [h for h in handlers if status.is_sleeping(body=cause.body, handler=h)]
handlers_todo = [h for h in handlers if status.is_awakened(body=cause.body, handler=h)]
