diff --git a/.gitignore b/.gitignore index 34a7e62..5097af5 100644 --- a/.gitignore +++ b/.gitignore @@ -169,3 +169,5 @@ tmp/ uv.lock .codspeed/ + +.vscode diff --git a/pyproject.toml b/pyproject.toml index ad43dbf..0929146 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,8 +4,8 @@ requires = ["setuptools>=42", "wheel"] [project] authors = [ - {name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru"}, - {name = "Dmitry Kutlubaev", email = "kutlubaev00@mail.ru"} + { name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru" }, + { name = "Dmitry Kutlubaev", email = "kutlubaev00@mail.ru" }, ] classifiers = [ "Development Status :: 4 - Beta", @@ -14,7 +14,7 @@ classifiers = [ "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", - "Programming Language :: Python :: 3.13" + "Programming Language :: Python :: 3.13", ] dependencies = [ "dataclass-wizard==0.*", @@ -24,14 +24,14 @@ dependencies = [ "pydantic==2.*", "sqlalchemy[asyncio]==2.0.*", "python-dotenv==1.*", - "typing-extensions>=4.0" + "typing-extensions>=4.0", ] description = "Event-Driven Architecture Framework for Distributed Systems" -maintainers = [{name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru"}] +maintainers = [{ name = "Vadim Kozyrevskiy", email = "vadikko2@mail.ru" }] name = "python-cqrs" readme = "README.md" requires-python = ">=3.10" -version = "4.11.0" +version = "4.12.0" [project.optional-dependencies] aiobreaker = ["aiobreaker>=0.3.0"] @@ -45,10 +45,10 @@ dev = [ "pytest-cov>=4.0.0", "pytest-codspeed==4.2.0", # Tests - "aio-pika==9.3.0", # from rabbit - "aiokafka==0.10.0", # from kafka - "retry-async==0.1.*", # from kafka - "requests==2.*", # from aiokafka + "aio-pika==9.3.0", # from rabbit + "aiokafka==0.10.0", # from kafka + "retry-async==0.1.*", # from kafka + "requests==2.*", # from aiokafka "pytest~=7.4.2", "pytest-asyncio~=0.21.1", "pytest-env==0.6.2", @@ -57,7 +57,7 @@ dev = [ "asyncpg>=0.29.0", "redis>=5.0.0", # Circuit breaker for tests - "aiobreaker>=0.3.0" # from aiobreaker + "aiobreaker>=0.3.0", # from aiobreaker ] examples = [ "fastapi==0.109.*", diff --git a/src/cqrs/events/event_emitter.py b/src/cqrs/events/event_emitter.py index e21af7a..f316fa3 100644 --- a/src/cqrs/events/event_emitter.py +++ b/src/cqrs/events/event_emitter.py @@ -1,3 +1,4 @@ +import asyncio import functools import logging import typing @@ -169,22 +170,32 @@ async def _(self, event: IDomainEvent) -> typing.Sequence[IEvent]: type(event).__name__, ) return () + + results = await asyncio.gather( + *(self._process_single_handler(event, item) for item in handlers_types), + return_exceptions=True, + ) + follow_ups: list[IEvent] = [] - for handler_item in handlers_types: - if isinstance(handler_item, EventHandlerFallback): - follow_ups.extend( - await self._handle_with_fallback(event, handler_item), + for handler_item, res in zip(handlers_types, results): + if isinstance(res, Exception): + handler_name = ( + handler_item.primary.__name__ + if isinstance(handler_item, EventHandlerFallback) + else handler_item.__name__ ) - continue - handler_type = handler_item - handler: _H = await self._container.resolve(handler_type) - logger.debug( - "Handling Event(%s) via event handler(%s)", - type(event).__name__, - handler_type.__name__, - ) - await handler.handle(event) - follow_ups.extend(list(handler.events)) + logger.exception( + "Error occurred while processing domain event %s in handler %s: %s", + type(event).__name__, + handler_name, + res, + exc_info=res, + ) + elif isinstance(res, BaseException): + raise res + else: + follow_ups.extend(res) + return follow_ups @emit.register(INotificationEvent) @@ -192,3 +203,22 @@ async def _(self, event: INotificationEvent) -> typing.Sequence[IEvent]: """Emit notification event: send to message broker; no follow-ups.""" await self._send_to_broker(event) return () + + async def _process_single_handler( + self, + event: IDomainEvent, + handler_item: typing.Union[typing.Type[_H], EventHandlerFallback], + ) -> typing.Sequence[IEvent]: + """Process a single event handler and return its follow-up events.""" + if isinstance(handler_item, EventHandlerFallback): + return await self._handle_with_fallback(event, handler_item) + + handler_type = handler_item + handler: _H = await self._container.resolve(handler_type) + logger.debug( + "Handling Event(%s) via event handler(%s)", + type(event).__name__, + handler_type.__name__, + ) + await handler.handle(event) + return list(handler.events) diff --git a/tests/benchmarks/dataclasses/test_benchmark_event_handler_chain.py b/tests/benchmarks/dataclasses/test_benchmark_event_handler_chain.py index feb6dbc..800df47 100644 --- a/tests/benchmarks/dataclasses/test_benchmark_event_handler_chain.py +++ b/tests/benchmarks/dataclasses/test_benchmark_event_handler_chain.py @@ -112,11 +112,15 @@ def test_benchmark_event_chain_three_levels_parallel( ) -> None: """Benchmark: 1 root event -> 10 L2 -> 50 L3 (61 total), semaphore 4 (dataclass).""" processor = event_processor_chain_parallel + loop = asyncio.new_event_loop() async def run() -> None: await processor.emit_events([_EventL1(id_="root")]) - benchmark(lambda: asyncio.run(run())) + try: + benchmark(lambda: loop.run_until_complete(run())) + finally: + loop.close() @pytest.fixture @@ -132,8 +136,12 @@ def test_benchmark_event_chain_three_levels_sequential( ) -> None: """Benchmark: same 3-level chain, sequential (BFS), dataclass events.""" processor = event_processor_chain_sequential + loop = asyncio.new_event_loop() async def run() -> None: await processor.emit_events([_EventL1(id_="root")]) - benchmark(lambda: asyncio.run(run())) + try: + benchmark(lambda: loop.run_until_complete(run())) + finally: + loop.close() diff --git a/tests/benchmarks/default/test_benchmark_event_handler_chain.py b/tests/benchmarks/default/test_benchmark_event_handler_chain.py index a335f54..f87b605 100644 --- a/tests/benchmarks/default/test_benchmark_event_handler_chain.py +++ b/tests/benchmarks/default/test_benchmark_event_handler_chain.py @@ -102,7 +102,7 @@ def _make_processor(parallel: bool) -> EventProcessor: @pytest.fixture def event_processor_chain_parallel() -> EventProcessor: - """EventProcessor with 3-level chain, parallel, semaphore=4.""" + """EventProcessor with 3-level chain, parallel, semaphore=4 (dataclass events).""" return _make_processor(parallel=True) @@ -111,18 +111,22 @@ def test_benchmark_event_chain_three_levels_parallel( benchmark, event_processor_chain_parallel: EventProcessor, ) -> None: - """Benchmark: 1 root event -> 10 L2 -> 50 L3 (61 total), semaphore 4.""" + """Benchmark: 1 root event -> 10 L2 -> 50 L3 (61 total), semaphore 4 (dataclass).""" processor = event_processor_chain_parallel + loop = asyncio.new_event_loop() async def run() -> None: await processor.emit_events([_EventL1(id="root")]) - benchmark(lambda: asyncio.run(run())) + try: + benchmark(lambda: loop.run_until_complete(run())) + finally: + loop.close() @pytest.fixture def event_processor_chain_sequential() -> EventProcessor: - """EventProcessor with 3-level chain, sequential.""" + """EventProcessor with 3-level chain, sequential (dataclass events).""" return _make_processor(parallel=False) @@ -131,10 +135,14 @@ def test_benchmark_event_chain_three_levels_sequential( benchmark, event_processor_chain_sequential: EventProcessor, ) -> None: - """Benchmark: same 3-level chain, sequential (BFS).""" + """Benchmark: same 3-level chain, sequential (BFS), dataclass events.""" processor = event_processor_chain_sequential + loop = asyncio.new_event_loop() async def run() -> None: await processor.emit_events([_EventL1(id="root")]) - benchmark(lambda: asyncio.run(run())) + try: + benchmark(lambda: loop.run_until_complete(run())) + finally: + loop.close() diff --git a/tests/unit/test_event_emitter_handler_events.py b/tests/unit/test_event_emitter_handler_events.py index 2904c94..fee1281 100644 --- a/tests/unit/test_event_emitter_handler_events.py +++ b/tests/unit/test_event_emitter_handler_events.py @@ -1,5 +1,6 @@ """Unit tests for EventEmitter returning follow-up events from handler.events.""" +import asyncio import typing import pydantic @@ -122,3 +123,107 @@ async def resolve(self, type_: type) -> EventHandler[IEvent]: assert len(follow_ups) == 2 xs = [e.x for e in follow_ups] # type: ignore[attr-defined] assert "a_1" in xs and "a_2" in xs + + +async def test_event_emitter_executes_handlers_in_parallel() -> None: + """Arrange: one event with two handlers that wait on each other. Act: emit in parallel. Assert: completes without deadlock.""" + + class _ParallelEvent(DomainEvent, frozen=True): + id: str + + handler_1_started = asyncio.Event() + handler_2_started = asyncio.Event() + handler_1_completed = asyncio.Event() + handler_2_completed = asyncio.Event() + + class _ParallelHandler1(EventHandler[_ParallelEvent]): + async def handle(self, event: _ParallelEvent) -> None: + handler_1_started.set() + await asyncio.wait_for(handler_2_started.wait(), timeout=1.0) + handler_1_completed.set() + + class _ParallelHandler2(EventHandler[_ParallelEvent]): + async def handle(self, event: _ParallelEvent) -> None: + handler_2_started.set() + await asyncio.wait_for(handler_1_started.wait(), timeout=1.0) + handler_2_completed.set() + + h1 = _ParallelHandler1() + h2 = _ParallelHandler2() + + event_map = EventMap() + event_map.bind(_ParallelEvent, _ParallelHandler1) + event_map.bind(_ParallelEvent, _ParallelHandler2) + + class C: + async def resolve(self, type_: type) -> EventHandler[IEvent]: + if type_ is _ParallelHandler1: + return h1 # type: ignore[return-value] + return h2 # type: ignore[return-value] + + emitter = EventEmitter( + event_map=event_map, + container=C(), # type: ignore[arg-type] + ) + + # If executed sequentially, this call will deadlock and raise asyncio.TimeoutError + await emitter.emit(_ParallelEvent(id="p1")) + + assert handler_1_started.is_set() + assert handler_2_started.is_set() + assert handler_1_completed.is_set() + assert handler_2_completed.is_set() + + +async def test_event_emitter_continues_execution_when_handler_raises_exception() -> None: + """Arrange: two handlers where handler 1 fails. Act: emit. Assert: handler 2 completes without flow crashing.""" + + class _SampleEvent(DomainEvent, frozen=True): + id: str + + handler_1_executed = asyncio.Event() + handler_2_executed = asyncio.Event() + + class _FailingHandler(EventHandler[_SampleEvent]): + async def handle(self, event: _SampleEvent) -> None: + handler_1_executed.set() + raise RuntimeError("Database connection lost") + + class _HealthyHandler(EventHandler[_SampleEvent]): + def __init__(self) -> None: + self._out: list[IEvent] = [] + + @property + def events(self) -> typing.Sequence[IEvent]: + return tuple(self._out) + + async def handle(self, event: _SampleEvent) -> None: + # Brief pause to ensure concurrency while handler 1 fails + await asyncio.sleep(0.01) + handler_2_executed.set() + self._out.append(_EventB(id="success_" + event.id)) + + failing_h = _FailingHandler() + healthy_h = _HealthyHandler() + + event_map = EventMap() + event_map.bind(_SampleEvent, _FailingHandler) + event_map.bind(_SampleEvent, _HealthyHandler) + + class Container: + async def resolve(self, type_: type) -> EventHandler[IEvent]: + if type_ is _FailingHandler: + return failing_h # type: ignore[return-value] + return healthy_h # type: ignore[return-value] + + emitter = EventEmitter( + event_map=event_map, + container=Container(), # type: ignore[arg-type] + ) + follow_ups = await emitter.emit(_SampleEvent(id="123")) + + assert handler_1_executed.is_set() + assert handler_2_executed.is_set() + assert len(follow_ups) == 1 + assert isinstance(follow_ups[0], _EventB) + assert follow_ups[0].id_ == "success_123" # type: ignore[attr-defined] diff --git a/tests/unit/test_event_fallback.py b/tests/unit/test_event_fallback.py index 2a18f0d..ad9f36f 100644 --- a/tests/unit/test_event_fallback.py +++ b/tests/unit/test_event_fallback.py @@ -102,8 +102,7 @@ async def test_event_fallback_failure_exceptions_only_matching_triggers_fallback container: Container[Any] = _TestEventContainer() emitter = EventEmitter(event_map=event_map, container=container) - with pytest.raises(RuntimeError, match="Primary failed"): - await emitter.emit(SampleEvent(id="e1")) + await emitter.emit(SampleEvent(id="e1")) assert container._primary.called assert not container._fallback.called