Choreographic Saga pattern supporting

Author: Вадим Козыревский

README.md

@@ -199,7 +199,7 @@ the [documentation](https://github.com/vadikko2/cqrs/blob/master/examples/cor_re
 
 ## Saga Pattern
 
-The package implements the Choreographic Saga pattern for managing distributed transactions across multiple services or operations. 
+The package implements the Choreographic Saga pattern for managing distributed transactions across multiple services or operations.
 Sagas enable eventual consistency by executing a series of steps where each step can be compensated if a subsequent step fails.
 
 ### Key Features
@@ -236,13 +236,13 @@ class ProcessPaymentResponse(Response):
 class ReserveInventoryStep(SagaStepHandler[OrderContext, ReserveInventoryResponse]):
     def __init__(self, inventory_service):
         self._inventory_service = inventory_service
-    
+
     async def act(self, context: OrderContext) -> SagaStepResult[OrderContext, ReserveInventoryResponse]:
         # Reserve inventory
         reservation_id = await self._inventory_service.reserve_items(context.order_id, context.items)
         context.inventory_reservation_id = reservation_id
         return self._generate_step_result(ReserveInventoryResponse(reservation_id=reservation_id))
-    
+
     async def compensate(self, context: OrderContext) -> None:
         # Release inventory if saga fails
         if context.inventory_reservation_id:
@@ -251,13 +251,13 @@ class ReserveInventoryStep(SagaStepHandler[OrderContext, ReserveInventoryRespons
 class ProcessPaymentStep(SagaStepHandler[OrderContext, ProcessPaymentResponse]):
     def __init__(self, payment_service):
         self._payment_service = payment_service
-    
+
     async def act(self, context: OrderContext) -> SagaStepResult[OrderContext, ProcessPaymentResponse]:
         # Process payment
         payment_id = await self._payment_service.charge(context.order_id, context.total_amount)
         context.payment_id = payment_id
         return self._generate_step_result(ProcessPaymentResponse(payment_id=payment_id))
-    
+
     async def compensate(self, context: OrderContext) -> None:
         # Refund payment if saga fails
         if context.payment_id:
@@ -281,8 +281,8 @@ async with saga.transaction(context=context, saga_id=saga_id) as transaction:
         # If any step fails, compensation happens automatically
 ```
 
-The saga state and step history are persisted to `SagaStorage`. The `SagaLog` maintains a complete audit trail 
-of all step executions (both `act` and `compensate` operations) with timestamps and status information. 
+The saga state and step history are persisted to `SagaStorage`. The `SagaLog` maintains a complete audit trail
+of all step executions (both `act` and `compensate` operations) with timestamps and status information.
 This enables the recovery mechanism to restore saga state and ensure eventual consistency even after system failures.
 
 If a saga is interrupted (e.g., due to a crash), you can recover it using the recovery mechanism:

examples/saga.py

@@ -92,7 +92,6 @@
 from cqrs.saga.models import SagaContext
 from cqrs.saga.saga import Saga
 from cqrs.saga.step import SagaStepHandler, SagaStepResult
-from cqrs.saga.storage.enums import SagaStatus
 from cqrs.saga.storage.memory import MemorySagaStorage
 
 logging.basicConfig(level=logging.INFO)

examples/saga_fastapi_sse.py

@@ -463,7 +463,9 @@ async def generate_sse():
             yield f"data: {json.dumps(start_data)}\n\n"
 
             # Execute saga with saga_id for persistence
-            async with saga.transaction(context=context, saga_id=saga_id) as transaction:
+            async with saga.transaction(
+                context=context, saga_id=saga_id
+            ) as transaction:
                 async for step_result in transaction:
                     completed_steps += 1
                     step_name = step_result.step_type.__name__

examples/saga_recovery.py

@@ -301,7 +301,9 @@ async def cancel_shipment(self, shipment_id: str) -> None:
 
         tracking_number = self._shipments[shipment_id]
         del self._shipments[shipment_id]
-        logger.info(f"  ↻ Cancelled shipment {shipment_id} (tracking: {tracking_number})")
+        logger.info(
+            f"  ↻ Cancelled shipment {shipment_id} (tracking: {tracking_number})"
+        )
 
 
 # ============================================================================
@@ -562,7 +564,9 @@ async def simulate_interrupted_saga() -> tuple[uuid.UUID, MemorySagaStorage]:
                 # Simulate crash after first step
                 if step_name == "ReserveInventoryStep":
                     print("\n💥 SIMULATED SERVER CRASH!")
-                    print("   (In reality: server restart, network failure, timeout, etc.)")
+                    print(
+                        "   (In reality: server restart, network failure, timeout, etc.)"
+                    )
                     print("   Saga state has been persisted to storage.")
                     print("   Current state: RUNNING, 1 step completed")
                     # Break out of loop to simulate interruption

src/cqrs/middlewares/base.py

@@ -29,8 +29,8 @@ def wrap(self, handle: HandleType) -> HandleType:
         return handle
 
 
-# Contravariant TypeVar for SagaMiddleware protocol
-SagaContextT = typing.TypeVar("SagaContextT", bound=SagaContext, contravariant=True)
+# TypeVar for SagaMiddleware protocol
+SagaContextT = typing.TypeVar("SagaContextT", bound=SagaContext)
 
 SagaHandlerType = typing.Callable[[SagaContextT], typing.Awaitable[None]]
 

src/cqrs/saga/models.py

@@ -1,6 +1,9 @@
 import dataclasses
 import typing
 
+# Type variable for from_dict classmethod return type
+_T = typing.TypeVar("_T", bound="SagaContext")
+
 
 @dataclasses.dataclass
 class SagaContext:
@@ -29,7 +32,7 @@ def to_dict(self) -> dict[str, typing.Any]:
         return dataclasses.asdict(self)
 
     @classmethod
-    def from_dict(cls, data: dict[str, typing.Any]) -> "SagaContext":
+    def from_dict(cls: type[_T], data: dict[str, typing.Any]) -> _T:
         """
         Deserialize context from dictionary.