Add streaming handlers supporting (#22)

🚀 New Features

Streaming Support & SSE

StreamingRequestMediator: Added a new mediator designed for handling streaming requests that yield results incrementally.
StreamingRequestHandler: Introduced a new handler type for processing large batches or long-running operations with real-time progress updates.
FastAPI SSE Integration: Added native support and examples for using StreamingRequestMediator with Server-Sent Events (SSE) in FastAPI applications.
Parallel Event Processing

Concurrent Event Handling: Both RequestMediator and StreamingRequestMediator now support processing domain events in parallel.
Concurrency Control: Added max_concurrent_event_handlers parameter to limit the number of simultaneously running event handlers.
Configuration: Added concurrent_event_handle_enable flag to toggle between sequential and parallel execution.
Dependency Injection

dependency-injector Support: Added explicit support and documentation for the dependency-injector library using the DependencyInjectorCQRSContainer adapter.
📚 Documentation

Updated README with comprehensive examples for:
Streaming Request Handlers.
Parallel Event Processing configuration.
FastAPI SSE implementation.
DI container setup (both di and dependency-injector).

Author: vadikko2

.pre-commit-config.yaml

@@ -3,15 +3,15 @@ repos:
   - id: check-toml
   - id: check-docstring-first
   - id: check-ast
-  - exclude: (^tests/mock/|^tests/integration/fixtures)
+  - exclude: (^tests/mock/|^tests/integration/|^tests/fixtures)
     id: trailing-whitespace
   - id: end-of-file-fixer
   - id: check-yaml
   - id: check-toml
   - id: check-added-large-files
   - args:
     - --pytest-test-first
-    exclude: (^tests/mock/|^tests/integration/fixtures)
+    exclude: (^tests/mock/|^tests/integration/|^tests/fixtures)
     id: name-tests-test
   - id: check-merge-conflict
   - id: check-json
@@ -22,7 +22,6 @@ repos:
   repo: https://github.com/asottile/add-trailing-comma
   rev: v3.1.0
 - hooks:
-  - id: pretty-format-ini
   - args:
     - --autofix
     - --indent
@@ -58,7 +57,7 @@ repos:
   repo: https://github.com/astral-sh/ruff-pre-commit
   rev: v0.6.1
 - repo: https://github.com/RobertCraigie/pyright-python
-  rev: v1.1.387
+  rev: v1.1.380
   hooks:
   - id: pyright
     types: [python]

README.md

@@ -18,7 +18,9 @@ project ([documentation](https://akhundmurad.github.io/diator/)) with several en
    that `Notification` and `ECST` events are sent to the broker;
 7. FastAPI supporting;
 8. FastStream supporting;
-9. [Protobuf](https://protobuf.dev/) events supporting.
+9. [Protobuf](https://protobuf.dev/) events supporting;
+10. `StreamingRequestMediator` and `StreamingRequestHandler` for handling streaming requests with real-time progress updates;
+11. Parallel event processing with configurable concurrency limits.
 
 ## Request Handlers
 
@@ -98,6 +100,40 @@ class ReadMeetingQueryHandler(RequestHandler[ReadMeetingQuery, ReadMeetingQueryR
 A complete example can be found in
 the [documentation](https://github.com/vadikko2/cqrs/blob/master/examples/request_handler.py)
 
+### Streaming Request Handler
+
+Streaming Request Handler processes requests incrementally and yields results as they become available.
+This is particularly useful for processing large batches of items, file uploads, or any operation that benefits from
+real-time progress updates.
+
+`StreamingRequestHandler` works with `StreamingRequestMediator` that streams results to clients in real-time.
+
+```python
+import typing
+from cqrs.requests.request_handler import StreamingRequestHandler
+from cqrs.events.event import Event
+
+class ProcessFilesCommandHandler(StreamingRequestHandler[ProcessFilesCommand, FileProcessedResult]):
+    def __init__(self):
+        self._events: list[Event] = []
+
+    @property
+    def events(self) -> list[Event]:
+        return self._events.copy()
+
+    async def handle(self, request: ProcessFilesCommand) -> typing.AsyncIterator[FileProcessedResult]:
+        for file_id in request.file_ids:
+            # Process file
+            result = FileProcessedResult(file_id=file_id, status="completed", ...)
+            # Emit events
+            self._events.append(FileProcessedEvent(file_id=file_id, ...))
+            yield result
+            self._events.clear()
+```
+
+A complete example can be found in
+the [documentation](https://github.com/vadikko2/cqrs/blob/master/examples/streaming_handler_parallel_events.py)
+
 ## Event Handlers
 
 Event handlers are designed to process `Notification` and `ECST` events that are consumed from the broker.
@@ -129,6 +165,36 @@ class UserJoinedEventHandler(cqrs.EventHandler[UserJoined]):
 A complete example can be found in
 the [documentation](https://github.com/vadikko2/cqrs/blob/master/examples/domain_event_handler.py)
 
+### Parallel Event Processing
+
+Both `RequestMediator` and `StreamingRequestMediator` support parallel processing of domain events. You can control
+the number of event handlers that run simultaneously using the `max_concurrent_event_handlers` parameter.
+
+This feature is especially useful when:
+- Multiple event handlers need to process events independently
+- You want to improve performance by processing events concurrently
+- You need to limit resource consumption by controlling concurrency
+
+**Configuration:**
+
+```python
+from cqrs.requests import bootstrap
+
+mediator = bootstrap.bootstrap_streaming(
+    di_container=container,
+    commands_mapper=commands_mapper,
+    domain_events_mapper=domain_events_mapper,
+    message_broker=broker,
+    max_concurrent_event_handlers=3,  # Process up to 3 events in parallel
+    concurrent_event_handle_enable=True,  # Enable parallel processing
+)
+```
+
+> [!TIP]
+> - Set `max_concurrent_event_handlers` to limit the number of simultaneously running event handlers
+> - Set `concurrent_event_handle_enable=False` to disable parallel processing and process events sequentially
+> - The default value for `max_concurrent_event_handlers` is `10` for `StreamingRequestMediator` and `1` for `RequestMediator`
+
 ## Producing Notification Events
 
 During the handling of a command, `cqrs.NotificationEvent` events may be generated and then sent to the broker.
@@ -317,6 +383,10 @@ The current version of the python-cqrs package does not support the implementati
 Use the following example to set up dependency injection in your command, query and event handlers. This will make
 dependency management simpler.
 
+The package supports two DI container libraries:
+
+### di library
+
 ```python
 import di
 ...
@@ -342,7 +412,35 @@ def setup_di() -> di.Container:
 ```
 
 A complete example can be found in
-the [documentation](https://github.com/vadikko2/python-cqrs/blob/master/examples/dependency_injection.py)
+the [documentation](https://github.com/vadikko2/cqrs/blob/master/examples/dependency_injection.py)
+
+### dependency-injector library
+
+The package also supports [dependency-injector](https://github.com/ets-labs/python-dependency-injector) library.
+You can use `DependencyInjectorCQRSContainer` adapter to integrate dependency-injector containers with python-cqrs.
+
+```python
+from dependency_injector import containers, providers
+from cqrs.container.dependency_injector import DependencyInjectorCQRSContainer
+
+class ApplicationContainer(containers.DeclarativeContainer):
+    # Define your providers
+    service = providers.Factory(ServiceImplementation)
+
+# Create CQRS container adapter
+cqrs_container = DependencyInjectorCQRSContainer(ApplicationContainer())
+
+# Use with bootstrap
+mediator = bootstrap.bootstrap(
+    di_container=cqrs_container,
+    commands_mapper=commands_mapper,
+    ...
+)
+```
+
+Complete examples can be found in:
+- [Simple example](https://github.com/vadikko2/cqrs/blob/master/examples/dependency_injector_integration_simple_example.py)
+- [Practical example with FastAPI](https://github.com/vadikko2/cqrs/blob/master/examples/dependency_injector_integration_practical_example.py)
 
 ## Mapping
 
@@ -485,6 +583,54 @@ async def hello_world_event_handler(
 A complete example can be found in
 the [documentation](https://github.com/vadikko2/python-cqrs/blob/master/examples/kafka_event_consuming.py)
 
+### FastAPI SSE Streaming
+
+`StreamingRequestMediator` is ready and designed for use with Server-Sent Events (SSE) in FastAPI applications.
+This allows you to stream results to clients in real-time as they are processed.
+
+**Example FastAPI endpoint with SSE:**
+
+```python
+import fastapi
+import json
+from cqrs.requests import bootstrap
+
+def streaming_mediator_factory() -> cqrs.StreamingRequestMediator:
+    return bootstrap.bootstrap_streaming(
+        di_container=container,
+        commands_mapper=commands_mapper,
+        domain_events_mapper=domain_events_mapper,
+        message_broker=broker,
+        max_concurrent_event_handlers=3,
+        concurrent_event_handle_enable=True,
+    )
+
+@app.post("/process-files")
+async def process_files_stream(
+    command: ProcessFilesCommand,
+    mediator: cqrs.StreamingRequestMediator = fastapi.Depends(streaming_mediator_factory),
+) -> fastapi.responses.StreamingResponse:
+    async def generate_sse():
+        yield f"data: {json.dumps({'type': 'start', 'message': 'Processing...'})}\n\n"
+
+        async for result in mediator.stream(command):
+            sse_data = {
+                "type": "progress",
+                "data": result.model_dump(),
+            }
+            yield f"data: {json.dumps(sse_data)}\n\n"
+
+        yield f"data: {json.dumps({'type': 'complete'})}\n\n"
+
+    return fastapi.responses.StreamingResponse(
+        generate_sse(),
+        media_type="text/event-stream",
+    )
+```
+
+A complete example can be found in
+the [documentation](https://github.com/vadikko2/cqrs/blob/master/examples/fastapi_sse_streaming.py)
+
 ## Protobuf messaging
 
 The `python-cqrs` package supports integration with [protobuf](https://developers.google.com/protocol-buffers/).\

examples/dependency_injection.py

@@ -1,3 +1,56 @@
+"""
+Example: Dependency Injection with CQRS Request Handlers
+
+This example demonstrates how to use dependency injection (DI) with CQRS request handlers.
+The system shows how to integrate DI containers with CQRS mediators, allowing handlers
+to receive dependencies through constructor injection.
+
+Use case: Decoupling business logic from infrastructure dependencies. Handlers can
+declare their dependencies in constructors, and the DI container automatically resolves
+and injects them when handlers are instantiated.
+
+================================================================================
+HOW TO RUN THIS EXAMPLE
+================================================================================
+
+Run the example:
+   python examples/dependency_injection.py
+
+The example will:
+- Set up a DI container with dependency bindings
+- Create a command handler that requires a dependency
+- Execute the command and verify the dependency was injected correctly
+
+================================================================================
+WHAT THIS EXAMPLE DEMONSTRATES
+================================================================================
+
+1. Dependency Injection Setup:
+   - Define abstract dependencies (AbstractDependency) and concrete implementations
+   - Configure DI container with type bindings
+   - Use scope="request" to create new instances per request
+
+2. Constructor Injection:
+   - Command handlers receive dependencies through constructor parameters
+   - Dependencies are automatically resolved by the DI container
+   - No need for manual instantiation or service locator pattern
+
+3. DI Container Integration:
+   - Pass DI container to bootstrap.bootstrap()
+   - Mediator uses the container to resolve handler dependencies
+   - Handlers are created with all required dependencies injected
+
+================================================================================
+REQUIREMENTS
+================================================================================
+
+Make sure you have installed:
+   - cqrs (this package)
+   - di (dependency injection library)
+
+================================================================================
+"""
+
 import abc
 import asyncio
 import logging

examples/domain_event_handler.py

@@ -1,3 +1,63 @@
+"""
+Example: Domain Event Handling
+
+This example demonstrates the basic pattern of domain event handling in CQRS.
+The system shows how command handlers emit domain events that are automatically
+dispatched to their corresponding event handlers.
+
+Use case: Separating command execution from side effects. When a command is executed,
+it emits domain events that represent what happened. These events are then processed
+by event handlers that perform side effects like sending notifications, updating
+read models, or triggering other workflows.
+
+================================================================================
+HOW TO RUN THIS EXAMPLE
+================================================================================
+
+Run the example:
+   python examples/domain_event_handler.py
+
+The example will:
+- Execute multiple JoinMeetingCommand commands
+- Emit UserJoined domain events for each command
+- Process events through UserJoinedEventHandler
+- Display the number of users in the meeting and events handled
+
+================================================================================
+WHAT THIS EXAMPLE DEMONSTRATES
+================================================================================
+
+1. Domain Event Definition:
+   - Create domain events as frozen dataclasses (UserJoined)
+   - Events represent something that happened in the domain
+   - Events are immutable and contain all relevant data
+
+2. Event Emission from Command Handlers:
+   - Command handlers collect events in the events property
+   - Events are emitted after command execution succeeds
+   - Multiple events can be emitted from a single command
+
+3. Event Handler Registration:
+   - Register event handlers using domain_events_mapper
+   - Map event types to their handlers
+   - Mediator automatically dispatches events to registered handlers
+
+4. Automatic Event Dispatching:
+   - Mediator collects events from command handlers
+   - Events are automatically sent to their registered handlers
+   - Event handlers process events asynchronously
+
+================================================================================
+REQUIREMENTS
+================================================================================
+
+Make sure you have installed:
+   - cqrs (this package)
+   - di (dependency injection)
+
+================================================================================
+"""
+
 import asyncio
 import logging
 import typing