refactor: side effect runners always run the side effect in the event loop provided to them regardless of the return value of the side effect being a coroutine or not, this is because even if the side effect is not a coroutine, it might still use async features internally

sassanh · sassanh · commit 75a865d16bd5 · 2025-02-27T04:28:42.000+04:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## Upcoming
 
 - refactor: provide correct signature for the autorun instance based on the function it decorates
+- refactor: side effect runners always run the side effect in the event loop provided to them regardless of the return value of the side effect being a coroutine or not, this is because even if the side effect is not a coroutine, it might still use async features internally
 
 ## Version 0.18.3
 
diff --git a/redux/main.py b/redux/main.py
@@ -93,7 +93,10 @@ def __init__(
             tuple[EventHandler[Event], Event] | None
         ]()
         self._workers = [
-            SideEffectRunnerThread(task_queue=self._event_handlers_queue)
+            SideEffectRunnerThread(
+                task_queue=self._event_handlers_queue,
+                create_task=self._create_task,
+            )
             for _ in range(self.store_options.threads)
         ]
         for worker in self._workers:
diff --git a/redux/side_effect_runner.py b/redux/side_effect_runner.py
@@ -4,14 +4,14 @@
 
 import asyncio
 import contextlib
+import inspect
 import threading
 import weakref
 from asyncio import Handle, iscoroutine
 from collections.abc import Callable
-from inspect import signature
 from typing import TYPE_CHECKING, Any, Generic, cast
 
-from redux.basic_types import Event, EventHandler
+from redux.basic_types import Event, EventHandler, TaskCreator
 
 if TYPE_CHECKING:
     import queue
@@ -24,15 +24,14 @@ def __init__(
         self: SideEffectRunnerThread,
         *,
         task_queue: queue.Queue[tuple[EventHandler[Event], Event] | None],
+        create_task: TaskCreator | None,
     ) -> None:
         """Initialize the side effect runner thread."""
         super().__init__()
         self.task_queue = task_queue
         self.loop = asyncio.get_event_loop()
         self._handles: set[Handle] = set()
-        self.create_task = lambda coro: self._handles.add(
-            self.loop.call_soon_threadsafe(self.loop.create_task, coro),
-        )
+        self.create_task = create_task
 
     def run(self: SideEffectRunnerThread[Event]) -> None:
         """Run the side effect runner thread."""
@@ -51,12 +50,27 @@ def run(self: SideEffectRunnerThread[Event]) -> None:
                     event_handler = event_handler_
                 parameters = 1
                 with contextlib.suppress(Exception):
-                    parameters = len(signature(event_handler).parameters)
-                if parameters == 1:
-                    result = cast(Callable[[Event], Any], event_handler)(event)
-                else:
-                    result = cast(Callable[[], Any], event_handler)()
-                if iscoroutine(result):
-                    self.create_task(result)
+                    parameters = len(inspect.signature(event_handler).parameters)
+
+                if self.create_task:
+
+                    async def _(
+                        event_handler: EventHandler[Event],
+                        event: Event,
+                        parameters: int,
+                    ) -> None:
+                        if parameters == 1:
+                            result = cast(Callable[[Event], Any], event_handler)(event)
+                        else:
+                            result = cast(Callable[[], Any], event_handler)()
+                        if iscoroutine(result):
+                            await result
+
+                    self.create_task(_(event_handler, event, parameters))
+                else:  # noqa: PLR5501
+                    if parameters == 1:
+                        cast(Callable[[Event], Any], event_handler)(event)
+                    else:
+                        cast(Callable[[], Any], event_handler)()
             finally:
                 self.task_queue.task_done()
diff --git a/redux_pytest/fixtures/event_loop.py b/redux_pytest/fixtures/event_loop.py
@@ -8,24 +8,37 @@
 import pytest
 
 if TYPE_CHECKING:
-    from collections.abc import Coroutine
+    from collections.abc import Callable, Coroutine
+
+    from redux.basic_types import TaskCreatorCallback
 
 
 class LoopThread(threading.Thread):
     def __init__(self: LoopThread) -> None:
         super().__init__()
         self.loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(self.loop)
 
     def run(self: LoopThread) -> None:
         self.loop.run_forever()
 
     def stop(self: LoopThread) -> None:
-        asyncio.set_event_loop(None)
         self.loop.call_soon_threadsafe(self.loop.stop)
 
-    def create_task(self: LoopThread, coro: Coroutine) -> None:
-        self.loop.call_soon_threadsafe(self.loop.create_task, coro)
+    def create_task(
+        self: LoopThread,
+        coro: Coroutine,
+        *,
+        callback: TaskCreatorCallback | None = None,
+    ) -> None:
+        def _(
+            coro: Coroutine,
+            callback: Callable[[asyncio.Task], None] | None = None,
+        ) -> None:
+            task = self.loop.create_task(coro)
+            if callback:
+                task.add_done_callback(callback)
+
+        self.loop.call_soon_threadsafe(_, coro, callback)
 
 
 @pytest.fixture
diff --git a/tests/test_async.py b/tests/test_async.py
@@ -11,6 +11,7 @@
 from redux.basic_types import (
     AutorunOptions,
     BaseAction,
+    BaseEvent,
     CompleteReducerResult,
     CreateStoreOptions,
     FinishAction,
@@ -22,7 +23,7 @@
 from redux.main import Store
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Coroutine
+    from collections.abc import Generator
 
     from redux_pytest.fixtures.event_loop import LoopThread
 
@@ -37,21 +38,28 @@ class StateType(Immutable):
 class IncrementAction(BaseAction): ...
 
 
+class IncrementEvent(BaseEvent):
+    post_value: int
+
+
 class SetMirroredValueAction(BaseAction):
     value: int
 
 
 def reducer(
     state: StateType | None,
     action: Action,
-) -> StateType | CompleteReducerResult[StateType, Action, FinishEvent]:
+) -> StateType | CompleteReducerResult[StateType, Action, IncrementEvent]:
     if state is None:
         if isinstance(action, InitAction):
             return StateType(value=0, mirrored_value=0)
         raise InitializationActionError(action)
 
     if isinstance(action, IncrementAction):
-        return replace(state, value=state.value + 1)
+        return CompleteReducerResult(
+            state=replace(state, value=state.value + 1),
+            events=[IncrementEvent(post_value=state.value + 1)],
+        )
     if isinstance(action, SetMirroredValueAction):
         return replace(state, mirrored_value=action.value)
     return state
@@ -62,25 +70,16 @@ def reducer(
 
 
 @pytest.fixture
-def store(event_loop: LoopThread) -> StoreType:
-    def _create_task_with_callback(
-        coro: Coroutine,
-        callback: Callable[[asyncio.Task], None] | None = None,
-    ) -> None:
-        def create_task_with_callback() -> None:
-            task = event_loop.loop.create_task(coro)
-            if callback:
-                callback(task)
-
-        event_loop.loop.call_soon_threadsafe(create_task_with_callback)
-
-    return Store(
+def store(event_loop: LoopThread) -> Generator[StoreType, None, None]:
+    store = Store(
         reducer,
         options=CreateStoreOptions(
             auto_init=True,
-            task_creator=_create_task_with_callback,
+            task_creator=event_loop.create_task,
         ),
     )
+    yield store
+    store.subscribe_event(FinishEvent, lambda: event_loop.stop())
 
 
 def dispatch_actions(store: StoreType) -> None:
@@ -90,7 +89,6 @@ def dispatch_actions(store: StoreType) -> None:
 
 def test_autorun(
     store: StoreType,
-    event_loop: LoopThread,
 ) -> None:
     @store.autorun(lambda state: state.value)
     async def sync_mirror(value: int) -> int:
@@ -107,16 +105,12 @@ async def sync_mirror(value: int) -> int:
     def _(mirrored_value: int) -> None:
         if mirrored_value < INCREMENTS:
             return
-        event_loop.stop()
         store.dispatch(FinishAction())
 
     dispatch_actions(store)
 
 
-def test_autorun_autoawait(
-    store: StoreType,
-    event_loop: LoopThread,
-) -> None:
+def test_autorun_autoawait(store: StoreType) -> None:
     @store.autorun(lambda state: state.value, options=AutorunOptions(auto_await=False))
     async def sync_mirror(value: int) -> int:
         store.dispatch(SetMirroredValueAction(value=value))
@@ -139,14 +133,10 @@ async def _(values: tuple[int, int]) -> None:
         elif value < INCREMENTS:
             store.dispatch(IncrementAction())
         else:
-            event_loop.stop()
             store.dispatch(FinishAction())
 
 
-def test_autorun_default_value(
-    store: StoreType,
-    event_loop: LoopThread,
-) -> None:
+def test_autorun_default_value(store: StoreType) -> None:
     @store.autorun(lambda state: state.value, options=AutorunOptions(default_value=5))
     async def _(value: int) -> int:
         store.dispatch(SetMirroredValueAction(value=value))
@@ -156,19 +146,16 @@ async def _(value: int) -> int:
         lambda state: state.mirrored_value,
         lambda state: state.mirrored_value >= INCREMENTS,
     )
-    def _(mirrored_value: int) -> None:
+    async def _(mirrored_value: int) -> None:
         if mirrored_value < INCREMENTS:
             return
-        event_loop.stop()
+        await asyncio.sleep(0.1)
         store.dispatch(FinishAction())
 
     dispatch_actions(store)
 
 
-def test_view(
-    store: StoreType,
-    event_loop: LoopThread,
-) -> None:
+def test_view(store: StoreType) -> None:
     calls = []
 
     @store.view(lambda state: state.value)
@@ -184,12 +171,11 @@ async def _(value: int) -> None:
         if value < INCREMENTS:
             store.dispatch(IncrementAction())
         else:
-            event_loop.stop()
             store.dispatch(FinishAction())
             assert calls == list(range(INCREMENTS + 1))
 
 
-def test_view_await(store: StoreType, event_loop: LoopThread) -> None:
+def test_view_await(store: StoreType) -> None:
     calls = []
 
     @store.view(lambda state: state.value)
@@ -208,15 +194,11 @@ async def _(value: int) -> None:
         if value < INCREMENTS:
             store.dispatch(IncrementAction())
         else:
-            event_loop.stop()
             store.dispatch(FinishAction())
             assert calls == list(range(INCREMENTS + 1))
 
 
-def test_view_with_args(
-    store: StoreType,
-    event_loop: LoopThread,
-) -> None:
+def test_view_with_args(store: StoreType) -> None:
     calls = []
 
     @store.view(lambda state: state.value)
@@ -231,15 +213,11 @@ async def _(value: int) -> None:
         if value < INCREMENTS:
             store.dispatch(IncrementAction())
         else:
-            event_loop.stop()
             store.dispatch(FinishAction())
             assert calls == [j for i in list(range(INCREMENTS + 1)) for j in [i] * 2]
 
 
-def test_view_with_default_value(
-    store: StoreType,
-    event_loop: LoopThread,
-) -> None:
+def test_view_with_default_value(store: StoreType) -> None:
     calls = []
 
     @store.view(lambda state: state.value, options=ViewOptions(default_value=5))
@@ -253,51 +231,39 @@ async def _(value: int) -> None:
         if value < INCREMENTS:
             store.dispatch(IncrementAction())
         else:
-            event_loop.stop()
             store.dispatch(FinishAction())
             assert calls == list(range(INCREMENTS + 1))
 
     store.dispatch(InitAction())
 
 
-def test_subscription(
-    store: StoreType,
-    event_loop: LoopThread,
-) -> None:
+def test_subscription(store: StoreType) -> None:
     async def render(state: StateType) -> None:
+        await asyncio.sleep(0.1)
         if state.value == INCREMENTS:
             unsubscribe()
             store.dispatch(FinishAction())
-            event_loop.stop()
 
     unsubscribe = store.subscribe(render)
 
     dispatch_actions(store)
 
 
-def test_event_subscription(
-    store: StoreType,
-    event_loop: LoopThread,
-) -> None:
-    async def finish() -> None:
+def test_event_subscription(store: StoreType) -> None:
+    async def handler(event: IncrementEvent) -> None:
         await asyncio.sleep(0.1)
-        event_loop.stop()
+        if event.post_value == INCREMENTS:
+            unsubscribe()
+            store.dispatch(FinishAction())
 
-    store.subscribe_event(FinishEvent, finish)
-    store.dispatch(FinishAction())
+    unsubscribe = store.subscribe_event(IncrementEvent, handler)
 
     dispatch_actions(store)
 
 
-def test_event_subscription_with_no_task_creator(event_loop: LoopThread) -> None:
+def test_event_subscription_with_no_task_creator() -> None:
     store = Store(
         reducer,
         options=CreateStoreOptions(auto_init=True),
     )
-
-    async def finish() -> None:
-        await asyncio.sleep(0.1)
-        event_loop.stop()
-
-    store.subscribe_event(FinishEvent, finish)
     store.dispatch(FinishAction())
diff --git a/tests/test_weakref.py b/tests/test_weakref.py
diff --git a/uv.lock b/uv.lock
