🍻 simplify logic

Author: RF-Tar-Railt

arclet/letoderea/core.py

@@ -6,7 +6,7 @@
 from collections import defaultdict
 from dataclasses import dataclass
 from operator import attrgetter
-from itertools import chain, groupby
+from itertools import chain
 from types import AsyncGeneratorType
 from typing import Any, TypeVar, overload, cast
 from collections.abc import Callable, Coroutine
@@ -46,10 +46,10 @@ async def gather(self, context: Contexts):
 exc_pub = define(ExceptionEvent, name="internal/exception")
 
 
-async def publish_exc_event(event: ExceptionEvent):
+def publish_exc_event(event: ExceptionEvent):
     scopes = [sp for sp in _scopes.values() if sp.available]
     subs = [slot for sp in scopes for slot in sp.subscribers if slot.publisher_id != "$backend"]
-    await dispatch(event, slots=subs)
+    return add_task(dispatch(event, slots=subs))
 
 
 async def compute(event: Any, scope: str | Scope | None = None, slots: Iterable[SubscriberSlot] | None = None, inherit_ctx: Contexts | None = None):
@@ -96,7 +96,7 @@ async def dispatch(event: Any, scope: str | Scope | None = None, slots: Iterable
                     return
                 if isinstance(event, ExceptionEvent):  # pragma: no cover
                     return
-                await publish_exc_event(ExceptionEvent(event, subs[_i], result))
+                publish_exc_event(ExceptionEvent(event, subs[_i], result))
             if isinstance(result, AsyncGeneratorType):  # pragma: no cover
                 async for res in result:
                     if result is None or result is STOP:
@@ -135,9 +135,9 @@ async def serial_exec_concurrent(subs: list[Subscriber], ctx: Contexts):
 
 async def serial(event: Any, scope: str | Scope | None = None, slots: Iterable[SubscriberSlot] | None = None, inherit_ctx: Contexts | None = None):
     grouped, context_map = await compute(event, scope, slots, inherit_ctx)
-    for (priority, pub_id) in grouped:
-        contexts = context_map[pub_id]
-        gene = serial_exec_concurrent(grouped[(priority, pub_id)], contexts)
+    for key, subs in grouped.items():
+        contexts = context_map[key[1]]
+        gene = serial_exec_concurrent(subs, contexts)
         async for subscriber, result in gene:
             if result is None or result is STOP:
                 continue
@@ -148,7 +148,7 @@ async def serial(event: Any, scope: str | Scope | None = None, slots: Iterable[S
                     return result.args[0]
                 if isinstance(event, ExceptionEvent):
                     return
-                await publish_exc_event(ExceptionEvent(event, subscriber, result))
+                publish_exc_event(ExceptionEvent(event, subscriber, result))
             elif isinstance(result, AsyncGeneratorType):  # pragma: no cover
                 async for res in result:
                     if res is None or res is STOP:
@@ -164,9 +164,9 @@ async def serial(event: Any, scope: str | Scope | None = None, slots: Iterable[S
 
 async def broadcast(event: Any, scope: str | Scope | None = None, slots: Iterable[SubscriberSlot] | None = None, inherit_ctx: Contexts | None = None, concurrent: bool = False):  # pragma: no cover
     grouped, context_map = await compute(event, scope, slots, inherit_ctx)
-    for (priority, pub_id) in grouped.keys():
-        contexts = context_map[pub_id]
-        async for subscriber, result in (serial_exec_concurrent(grouped[(priority, pub_id)], contexts) if concurrent else serial_exec(grouped[(priority, pub_id)], contexts)):
+    for key, subs in grouped.items():
+        contexts = context_map[key[1]]
+        async for subscriber, result in (serial_exec_concurrent(subs, contexts) if concurrent else serial_exec(subs, contexts)):
             if result is None or result is STOP:
                 continue
             if result is BLOCK:
@@ -179,7 +179,7 @@ async def broadcast(event: Any, scope: str | Scope | None = None, slots: Iterabl
                     continue
                 if isinstance(event, ExceptionEvent):
                     return
-                await publish_exc_event(ExceptionEvent(event, subscriber, result))
+                publish_exc_event(ExceptionEvent(event, subscriber, result))
             elif isinstance(result, AsyncGeneratorType):
                 async for res in result:
                     if res is None or res is STOP:
@@ -200,7 +200,7 @@ async def _post(event: Any, scope: str | Scope | None = None, inherit_ctx: Conte
     res = await serial(event, scope, inherit_ctx=inherit_ctx)
     if res is None:
         return
-    if res.__class__ is Force:
+    if res.__class__ is Force:  # pragma: no cover
         res = res.value
     if validate and hasattr(event, "check_result"):
         return cast(Resultable, event).check_result(res.value if isinstance(res, Result) else res)

arclet/letoderea/scope.pyi

@@ -21,7 +21,7 @@ scope_ctx: ContextModel[Scope]
 global_propagators: list[Propagator]
 
 
-@dataclass
+@dataclass(slots=True, frozen=True)
 class SubscriberSlot:
     subscriber: Subscriber
     publisher_id: str

arclet/letoderea/subscriber.py

@@ -78,16 +78,15 @@ async def __call__(self, context: Contexts):
         cache[self.target] = fut = asyncio.Future()
         try:
             res = await self.sub.handle(context.copy(), inner=True)
-        except Exception as e:  # pragma: no cover
-            fut.set_exception(e)
-            fut.cancel()
-            raise
         except BaseException as e:  # pragma: no cover
             fut.set_exception(e)
             fut.cancel()
-            cache.pop(self.target, None)
+            if isinstance(e, Exception):
+                cache.pop(self.target, None)
             raise
         else:
+            if isinstance(res, _ExitException):
+                raise
             fut.set_result(res)
             return res
 
@@ -115,6 +114,8 @@ class CompileParam:
     __slots__ = ("name", "annotation", "default", "providers", "depend", "record")
 
     async def solve(self, context: Contexts | dict[str, Any]):
+        if self.depend:
+            return await self.depend(context)  # type: ignore
         if self.name in context:
             return context[self.name]
         if self.record and (res := await self.record(context)) is not None:  # type: ignore
@@ -219,6 +220,7 @@ def __init__(self, callable_target: Callable[..., R], *, priority: int = 16, pro
         self._propagates: list[Subscriber] = []
         self._propagator_cache: WeakSet[Propagator] = WeakSet()
         self._cursor = 0
+        self._after_propagates = 0
         self._listen = _listen
 
         if hasattr(callable_target, "__providers__"):
@@ -304,31 +306,21 @@ async def handle(self, context: Contexts, inner=False):
             context[STACK] = AsyncExitStack()
         try:
             if self._cursor:
-                _res = await self._run_propagate(context, self._propagates[: self._cursor])
-                if _res is STOP or _res is BLOCK:
-                    return _res
-            arguments: Contexts = {}  # type: ignore
+                await self._run_propagate(context, self._propagates[: self._cursor])
+            arguments = {}  # type: ignore
             for param in self.params:
-                if param.depend:
-                    dep_res = await param.depend(context)
-                    if dep_res is STOP or dep_res is BLOCK:
-                        return dep_res
-                    arguments[param.name] = dep_res
-                else:
-                    arguments[param.name] = await param.solve(context)
+                arguments[param.name] = await param.solve(context)
             if self.is_cm:
                 stack: AsyncExitStack = context[STACK]
                 result = await stack.enter_async_context(self._callable_target(**arguments))
             elif self.is_agen:
                 result = self._callable_target(**arguments)
             else:
                 result = await self._callable_target(**arguments)
-            if self._propagates:
+            if self._after_propagates:
                 context[RESULT] = result
                 propagate_result = await self._run_propagate(context, self._propagates[self._cursor :])
                 result = result if propagate_result is None else propagate_result
-                if result is STOP or result is BLOCK:
-                    return result
         except InnerHandlerException as e:
             if inner:
                 raise
@@ -369,9 +361,7 @@ async def _run_propagate(self, context: Contexts, propagates: list[Subscriber]):
                 else:
                     raise
             else:
-                if isinstance(result, ExitState):
-                    return result
-                if isinstance(result, _ExitException):  # pragma: no cover
+                if isinstance(result, _ExitException):
                     raise result
                 if isinstance(result, dict):
                     context.update(result)
@@ -384,7 +374,7 @@ async def _run_propagate(self, context: Contexts, propagates: list[Subscriber]):
                         await self._run_propagate(context, [x[0] for x in pending.pop(key)])
         if pending:
             key, (slot, *_) = pending.popitem()
-            raise ExceptionHandler.call(slot[1], slot[0].callable_target,context, inner=True)
+            raise ExceptionHandler.call(slot[1], slot[0].callable_target, context, inner=True)
         return context.get(RESULT)
 
     @overload
@@ -419,36 +409,27 @@ def propagate(self, func: TTarget[Any] | Propagator | None = None, *, prepend: b
             self._propagator_cache.add(func)
             return lambda: ([dispose() for dispose in disposes] or self._propagator_cache.discard(func))
 
-        def _dispose(x: Subscriber):
-            self._propagates.remove(x)
-            self._cursor -= 1
-
         def wrapper(callable_target: TTarget[Any], /):
             if isinstance(callable_target, Subscriber):
                 raise ValueError("Subscriber can't be propagated")
             _providers = [*(providers or []), *self.providers]
             if prepend:
-                sub = Subscriber(
-                    callable_target,
-                    priority=priority,
-                    providers=_providers,
-                    dispose=_dispose,
-                    once=once,
-                    _listen=self._listen,
-                )
+                def _dispose(x: Subscriber):
+                    self._propagates.remove(x)
+                    self._cursor -= 1
+
+                sub = Subscriber(callable_target, priority=priority, providers=_providers, dispose=_dispose, once=once, _listen=self._listen)
                 self._propagates.insert(self._cursor, sub)
                 self._cursor += 1
             else:
+                def _dispose(x: Subscriber):
+                    self._propagates.remove(x)
+                    self._after_propagates -= 1
+
                 _providers.append(ResultProvider())
-                sub = Subscriber(
-                    callable_target,
-                    priority=priority,
-                    providers=_providers,
-                    dispose=lambda x: self._propagates.remove(x),
-                    once=once,
-                    _listen=self._listen,
-                )
+                sub = Subscriber(callable_target, priority=priority, providers=_providers, dispose=_dispose, once=once, _listen=self._listen)
                 self._propagates.append(sub)
+                self._after_propagates += 1
             return sub.dispose
 
         if func:

tests/test_exception.py

@@ -1,3 +1,5 @@
+import asyncio
+
 import pytest
 
 import arclet.letoderea as le
@@ -26,5 +28,5 @@ async def _(event: le.ExceptionEvent, origin, exc: Exception, subscriber):
         executed.append(1)
 
     await le.publish(TestExcEvent("1"))
-
+    await asyncio.sleep(0)
     assert len(executed) == 3