"""Response for each processed order."""

order_id: str = pydantic.Field()

status: str = pydantic.Field()

processed_at: datetime = pydantic.Field()

"""Event emitted when an order is processed."""

order_id: str

customer_id: str

total_amount: float

items_count: int

"""Event for analytics tracking."""

order_id: str

customer_id: str

total_amount: float

"""Event for inventory updates."""

order_id: str

"""Event for audit logging."""

order_id: str

action: str

timestamp: datetime

cqrs.StreamingRequestHandler[ProcessOrdersCommand, OrderProcessedResult],

):

"""

def __init__(self):

self._events: list[cqrs.Event] = []

@property

def events(self) -> list[cqrs.Event]:

"""Return a copy of events collected during processing."""

return self._events.copy()

def clear_events(self) -> None:

"""Clear events after they have been processed and emitted."""

self._events.clear()

async def handle(

self,

request: ProcessOrdersCommand,

) -> typing.AsyncIterator[OrderProcessedResult]:

"""

logger.info(f"Starting to process {len(request.order_ids)} orders")

for order_id in request.order_ids:

# Simulate order processing (e.g., validate, calculate totals, etc.)

await asyncio.sleep(0.1) # Simulate processing time

# Mock order data

order_data = {

"order_id": order_id,

"customer_id": f"customer_{order_id[-1]}",

"total_amount": 100.0 + float(order_id[-1]) * 10,

"items": [{"product_id": f"prod_{i}", "quantity": i + 1} for i in range(3)],

"category": "electronics" if int(order_id[-1]) % 2 == 0 else "clothing",

}

ORDER_STORAGE[order_id] = order_data

# Create result

result = OrderProcessedResult(

order_id=order_id,

status="processed",

processed_at=datetime.now(),

items_count=len(order_data["items"]),

)

# Emit multiple domain events that will be processed in parallel

self._events.append(

OrderProcessedEvent(

order_id=order_id,

customer_id=order_data["customer_id"],

total_amount=order_data["total_amount"],

items_count=len(order_data["items"]),

processed_at=datetime.now(),

),

)

self._events.append(

OrderAnalyticsEvent(

order_id=order_id,

customer_id=order_data["customer_id"],

total_amount=order_data["total_amount"],

category=order_data["category"],

),

)

self._events.append(

InventoryUpdateEvent(

order_id=order_id,

items=[

{"product_id": item["product_id"], "quantity": item["quantity"]} for item in order_data["items"]

],

),

)

self._events.append(

AuditLogEvent(

order_id=order_id,

action="order_processed",

timestamp=datetime.now(),

metadata={"total_amount": order_data["total_amount"]},

),

)

logger.info(

f"Processed order {order_id}, emitted {len(self._events)} events",

)

"""

async def handle(self, event: OrderProcessedEvent) -> None:

"""Send email notification to customer."""

# Simulate email sending (network call)

await asyncio.sleep(0.05) # Simulate network delay

email_data = {

"to": f"customer_{event.customer_id}@example.com",

"subject": f"Order {event.order_id} Processed",

"body": f"Your order for ${event.total_amount} has been processed.",

"sent_at": datetime.now(),

}

EMAIL_SENT_LOG.append(email_data)

logger.info(

f"📧 Email sent for order {event.order_id} " f"to customer {event.customer_id}",

"""

async def handle(self, event: OrderAnalyticsEvent) -> None:

"""Update analytics counters."""

# Simulate analytics processing (database update)

await asyncio.sleep(0.03) # Simulate database delay

ANALYTICS_STORAGE[f"{event.category}_orders"] += 1

ANALYTICS_STORAGE[f"{event.category}_revenue"] += int(event.total_amount)

ANALYTICS_STORAGE["total_orders"] += 1

logger.info(

f"📊 Analytics updated for order {event.order_id} " f"in category {event.category}",

"""

async def handle(self, event: InventoryUpdateEvent) -> None:

"""Update inventory levels."""

# Simulate inventory update (database transaction)

await asyncio.sleep(0.04) # Simulate database delay

for item in event.items:

product_id: str = item["product_id"] # type: ignore[assignment]

quantity: int = item["quantity"] # type: ignore[assignment]

INVENTORY_STORAGE[product_id] -= quantity

logger.info(

f"📦 Inventory updated for order {event.order_id}, " f"items: {len(event.items)}",

"""

async def handle(self, event: AuditLogEvent) -> None:

"""Write audit log entry."""

# Simulate audit log writing (file/database write)

await asyncio.sleep(0.02) # Simulate I/O delay

audit_entry = {

"order_id": event.order_id,

"action": event.action,

"timestamp": event.timestamp.isoformat(),

"metadata": event.metadata,

}

AUDIT_LOG.append(audit_entry)

"""Map domain events to handlers."""

mapper.bind(OrderProcessedEvent, OrderProcessedEventHandler)

mapper.bind(OrderAnalyticsEvent, OrderAnalyticsEventHandler)

mapper.bind(InventoryUpdateEvent, InventoryUpdateEventHandler)

"""

logger.info("=" * 80)

logger.info("Streaming Handler with Parallel Event Processing Example")

logger.info("=" * 80)

# Create mediator with parallel event processing enabled

# max_concurrent_event_handlers=3 means up to 3 event handlers can run simultaneously

mediator = bootstrap.bootstrap_streaming(

di_container=di.Container(),

commands_mapper=commands_mapper,

domain_events_mapper=domain_events_mapper,

message_broker=devnull.DevnullMessageBroker(),

max_concurrent_event_handlers=3, # Process up to 3 events in parallel

concurrent_event_handle_enable=True, # Enable parallel processing

)

# Process a batch of orders

order_ids = ["order_001", "order_002", "order_003", "order_004", "order_005"]

command = ProcessOrdersCommand(order_ids=order_ids)

logger.info(f"\nProcessing {len(order_ids)} orders...")

logger.info("Each order will trigger 4 events processed in parallel\n")

start_time = datetime.now()

results = []

# Stream results as orders are processed

async for result in mediator.stream(command):

if result is not None and isinstance(result, OrderProcessedResult):

results.append(result)

logger.info(

f"✅ Received result: Order {result.order_id} "

f"({result.items_count} items) - Status: {result.status}",

)

end_time = datetime.now()

processing_time = (end_time - start_time).total_seconds()

# Allow fire-and-forget parallel event handlers to finish (EventProcessor

# uses create_task when concurrent_event_handle_enable=True and does not await)

await asyncio.sleep(0.3)

# Print summary

logger.info("\n" + "=" * 80)

logger.info("Processing Summary")

logger.info("=" * 80)

logger.info(f"Total orders processed: {len(results)}")

logger.info(f"Total processing time: {processing_time:.2f} seconds")

logger.info(f"Emails sent: {len(EMAIL_SENT_LOG)}")

logger.info(f"Analytics entries: {len(ANALYTICS_STORAGE)}")

logger.info(f"Audit log entries: {len(AUDIT_LOG)}")

logger.info("\nAnalytics breakdown:")

for key, value in sorted(ANALYTICS_STORAGE.items()):

logger.info(f" {key}: {value}")

logger.info("\nInventory levels:")

for product_id, quantity in sorted(INVENTORY_STORAGE.items()):

logger.info(f" {product_id}: {quantity} units")

logger.info("\n" + "=" * 80)

logger.info("Example completed successfully!")

logger.info("=" * 80)

# Verify results: one event-handler invocation per order per handler type

assert len(results) == len(order_ids)

assert len(EMAIL_SENT_LOG) == len(order_ids)

assert len(AUDIT_LOG) == len(order_ids)