Skip to main content Skills Marketplace Découvrez et explorez les compétences IA créées par la communauté.
Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Afficher les détails du prompt Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
npx skills add https://github.com/Dicklesworthstone/pi_agent_rust --skill saga-orchestrationLa commande reste sur une seule ligne. Faites défiler horizontalement pour la vérifier avant de la copier.
Vous préférez une copie locale ? Téléchargez les fichiers actuellement disponibles dans SkillsMP.
Télécharger Zip Téléchargement... Métiers associés SOC
Basé sur la classification professionnelle SOC
name saga-orchestration description Implement saga patterns for distributed transactions and cross-aggregate workflows. Use when coordinating multi-step business processes, handling compensating transactions, or managing long-running workflows.
Saga Orchestration
Patterns for managing distributed transactions and long-running business processes.
When to Use This Skill
Coordinating multi-service transactions
Implementing compensating transactions
Managing long-running business workflows
Handling failures in distributed systems
Building order fulfillment processes
Implementing approval workflows
Core Concepts
1. Saga Types
Choreography Orchestration
┌─────┐ ┌─────┐ ┌─────┐ ┌─────────────┐
│Svc A│─►│Svc B│─►│Svc C│ │ Orchestrator│
└─────┘ └─────┘ └─────┘ └──────┬──────┘
│ │ │ │
▼ ▼ ▼ ┌─────┼─────┐
Event Event Event ▼ ▼ ▼
┌────┐┌────┐┌────┐
│Svc1││Svc2││Svc3│
└────┘└────┘└────┘
2. Saga Execution States
State Description Started Saga initiated Pending Waiting for step completion Compensating Rolling back due to failure Completed All steps succeeded Failed Saga failed after compensation
Templates
Template 1: Saga Orchestrator Base
from
import
from
import
from
import
from
import
List
Dict
Any
Optional
from
import
import
class
SagaState
Enum
"started"
"pending"
"compensating"
"completed"
"failed"
@dataclass
class
SagaStep
str
str
str
str
"pending"
Optional
Dict
None
Optional
str
None
Optional
None
Optional
None
@dataclass
class
Saga
str
str
Dict
str
Any
List
int
0
class
SagaOrchestrator
ABC
"""Base class for saga orchestrators."""
def
__init__
self, saga_store, event_publisher
self
self
@abstractmethod
def
define_steps
self, data: Dict
List
"""Define the saga steps."""
pass
@property
@abstractmethod
def
saga_type
self
str
"""Unique saga type identifier."""
pass
async
def
start
self, data: Dict
"""Start a new saga."""
str
self
self
await
self
await
self
return
async
def
handle_step_completed
self, saga_id: str , step_name: str , result: Dict
"""Handle successful step completion."""
await
self
for
in
if
"completed"
break
1
if
len
await
self
await
self
else
await
self
await
self
async
def
handle_step_failed
self, saga_id: str , step_name: str , error: str
"""Handle step failure - start compensation."""
await
self
for
in
if
"failed"
break
await
self
await
self
async
def
_execute_next_step
self, saga: Saga
"""Execute the next step in the saga."""
if
len
return
"executing"
await
self
await
self
"saga_id"
"step_name"
async
def
_compensate
self, saga: Saga
"""Execute compensation for completed steps."""
for
in
range
1
1
1
if
"completed"
"compensating"
await
self
await
self
"saga_id"
"step_name"
"original_result"
async
def
handle_compensation_completed
self, saga_id: str , step_name: str
"""Handle compensation completion."""
await
self
for
in
if
"compensated"
break
all
in
"compensated"
"pending"
"failed"
for
in
if
await
self
await
self
async
def
_on_saga_completed
self, saga: Saga
"""Called when saga completes successfully."""
await
self
f"{self.saga_type} Completed"
"saga_id"
async
def
_on_saga_failed
self, saga: Saga
"""Called when saga fails after compensation."""
await
self
f"{self.saga_type} Failed"
"saga_id"
"error"
"Saga failed"
Template 2: Order Fulfillment Saga class OrderFulfillmentSaga (SagaOrchestrator ):
"""Orchestrates order fulfillment across services."""
@property
def saga_type (self ) -> str :
return "OrderFulfillment"
def define_steps (self, data: Dict ) -> List [SagaStep]:
return [
SagaStep(
name="reserve_inventory" ,
action="InventoryService.ReserveItems" ,
compensation="InventoryService.ReleaseReservation"
),
SagaStep(
name="process_payment" ,
action="PaymentService.ProcessPayment" ,
compensation="PaymentService.RefundPayment"
),
SagaStep(
name="create_shipment" ,
action="ShippingService.CreateShipment" ,
compensation="ShippingService.CancelShipment"
),
SagaStep(
name="send_confirmation" ,
action="NotificationService.SendOrderConfirmation" ,
compensation="NotificationService.SendCancellationNotice"
)
]
async def create_order (order_data: Dict ):
saga = OrderFulfillmentSaga(saga_store, event_publisher)
return await saga.start({
"order_id" : order_data["order_id" ],
"customer_id" : order_data["customer_id" ],
"items" : order_data["items" ],
"payment_method" : order_data["payment_method" ],
"shipping_address" : order_data["shipping_address" ]
})
class InventoryService :
async def handle_reserve_items (self, command: Dict ):
try :
reservation = await self .reserve(
command["items" ],
command["order_id" ]
)
await self .event_publisher.publish(
"SagaStepCompleted" ,
{
"saga_id" : command["saga_id" ],
"step_name" : "reserve_inventory" ,
"result" : {"reservation_id" : reservation.id }
}
)
except InsufficientInventoryError as e:
await self .event_publisher.publish(
"SagaStepFailed" ,
{
"saga_id" : command["saga_id" ],
"step_name" : "reserve_inventory" ,
"error" : str (e)
}
)
async def handle_release_reservation (self, command: Dict ):
await self .release_reservation(
command["original_result" ]["reservation_id" ]
)
await self .event_publisher.publish(
"SagaCompensationCompleted" ,
{
"saga_id" : command["saga_id" ],
"step_name" : "reserve_inventory"
}
)
Template 3: Choreography-Based Saga from dataclasses import dataclass
from typing import Dict , Any
import asyncio
@dataclass
class SagaContext :
"""Passed through choreographed saga events."""
saga_id: str
step: int
data: Dict [str , Any ]
completed_steps: list
class OrderChoreographySaga :
"""Choreography-based saga using events."""
def __init__ (self, event_bus ):
self .event_bus = event_bus
self ._register_handlers()
def _register_handlers (self ):
self .event_bus.subscribe("OrderCreated" , self ._on_order_created)
self .event_bus.subscribe("InventoryReserved" , self ._on_inventory_reserved)
self .event_bus.subscribe("PaymentProcessed" , self ._on_payment_processed)
self .event_bus.subscribe("ShipmentCreated" , self ._on_shipment_created)
self .event_bus.subscribe("PaymentFailed" , self ._on_payment_failed)
self .event_bus.subscribe("ShipmentFailed" , self ._on_shipment_failed)
async def _on_order_created (self, event: Dict ):
"""Step 1: Order created, reserve inventory."""
await self .event_bus.publish("ReserveInventory" , {
"saga_id" : event["order_id" ],
"order_id" : event["order_id" ],
"items" : event["items" ]
})
async def _on_inventory_reserved (self, event: Dict ):
"""Step 2: Inventory reserved, process payment."""
await self .event_bus.publish("ProcessPayment" , {
"saga_id" : event["saga_id" ],
"order_id" : event["order_id" ],
"amount" : event["total_amount" ],
"reservation_id" : event["reservation_id" ]
})
async def _on_payment_processed (self, event: Dict ):
"""Step 3: Payment done, create shipment."""
await self .event_bus.publish("CreateShipment" , {
"saga_id" : event["saga_id" ],
"order_id" : event["order_id" ],
"payment_id" : event["payment_id" ]
})
async def _on_shipment_created (self, event: Dict ):
"""Step 4: Complete - send confirmation."""
await self .event_bus.publish("OrderFulfilled" , {
"saga_id" : event["saga_id" ],
"order_id" : event["order_id" ],
"tracking_number" : event["tracking_number" ]
})
async def _on_payment_failed (self, event: Dict ):
"""Payment failed - release inventory."""
await self .event_bus.publish("ReleaseInventory" , {
"saga_id" : event["saga_id" ],
"reservation_id" : event["reservation_id" ]
})
await self .event_bus.publish("OrderFailed" , {
"order_id" : event["order_id" ],
"reason" : "Payment failed"
})
async def _on_shipment_failed (self, event: Dict ):
"""Shipment failed - refund payment and release inventory."""
await self .event_bus.publish("RefundPayment" , {
"saga_id" : event["saga_id" ],
"payment_id" : event["payment_id" ]
})
await self .event_bus.publish("ReleaseInventory" , {
"saga_id" : event["saga_id" ],
"reservation_id" : event["reservation_id" ]
})
Template 4: Saga with Timeouts class TimeoutSagaOrchestrator (SagaOrchestrator ):
"""Saga orchestrator with step timeouts."""
def __init__ (self, saga_store, event_publisher, scheduler ):
super ().__init__(saga_store, event_publisher)
self .scheduler = scheduler
async def _execute_next_step (self, saga: Saga ):
if saga.current_step >= len (saga.steps):
return
step = saga.steps[saga.current_step]
step.status = "executing"
step.timeout_at = datetime.utcnow() + timedelta(minutes=5 )
await self .saga_store.save(saga)
await self .scheduler.schedule(
f"saga_timeout_{saga.saga_id} _{step.name} " ,
self ._check_timeout,
{"saga_id" : saga.saga_id, "step_name" : step.name},
run_at=step.timeout_at
)
await self .event_publisher.publish(
step.action,
{"saga_id" : saga.saga_id, "step_name" : step.name, **saga.data}
)
async def _check_timeout (self, data: Dict ):
"""Check if step has timed out."""
saga = await self .saga_store.get(data["saga_id" ])
step = next (s for s in saga.steps if s.name == data["step_name" ])
if step.status == "executing" :
await self .handle_step_failed(
data["saga_id" ],
data["step_name" ],
"Step timed out"
)
Best Practices
Do's
Make steps idempotent - Safe to retryDesign compensations carefully - They must workUse correlation IDs - For tracing across servicesImplement timeouts - Don't wait foreverLog everything - For debugging failures
Don'ts
Don't assume instant completion - Sagas take timeDon't skip compensation testing - Most critical partDon't couple services - Use async messagingDon't ignore partial failures - Handle gracefully
Resources 