qpanda3_runtime.toolbox.failure_orchestrator#

Failure orchestrator module for managing and coordinating failure handling.

This module provides components for orchestrating failure handling strategies in distributed quantum computing systems. It includes mechanisms for detecting, analyzing, and responding to failures in a coordinated manner.

The failure orchestrator works in conjunction with the tracking system to provide comprehensive failure management capabilities.

Modules#

failure_orchestrator : Main orchestrator class and related components.

Examples

>>> from qpanda3_runtime.toolbox.failure_orchestrator import FailureOrchestrator
>>> orchestrator = FailureOrchestrator()
>>> orchestrator.register_handler("timeout", timeout_handler)
>>> orchestrator.handle_failure("timeout", context_data)

Notes

This module is designed to be extensible, allowing custom failure handlers to be registered for different types of failures.

Submodules#

qpanda3_runtime.toolbox.failure_orchestrator.failure_orchestrator

Classes#

`FailureOrchestrator`	Main orchestrator for coordinating failure handling.
`FailureHandler`	Abstract base class for failure handlers.
`FailureContext`	Context information for a failure event.
`FailureSeverity`	Severity levels for failure classification.
`OrchestrationStrategy`	Strategies for orchestrating failure handling.

Package Contents#

class qpanda3_runtime.toolbox.failure_orchestrator.FailureOrchestrator(default_strategy: OrchestrationStrategy = OrchestrationStrategy.IMMEDIATE)#

Main orchestrator for coordinating failure handling.

This class manages a collection of failure handlers and coordinates the handling of failures based on configured strategies.

Variables:: default_strategy (OrchestrationStrategy) – Default strategy to use when no specific strategy is configured.

Examples

>>> orchestrator = FailureOrchestrator()
>>> orchestrator.register_handler("timeout", TimeoutHandler())
>>> orchestrator.set_strategy("timeout", OrchestrationStrategy.IMMEDIATE)
>>>
>>> context = FailureContext(...)
>>> result = orchestrator.handle_failure(context)

register_handler(failure_type: str, handler: FailureHandler) → None#

Parameters:

failure_type (str) – Type identifier for the failure.
handler (FailureHandler) – Handler instance to register.

Raises:

ValueError – If the failure_type is empty or handler is None.

Notes

If a handler is already registered for the failure_type, it will be replaced with the new handler.

set_strategy(failure_type: str, strategy: OrchestrationStrategy) → None#

Set orchestration strategy for a specific failure type.

Parameters:

failure_type (str) – Type identifier for the failure.
strategy (OrchestrationStrategy) – Strategy to use for orchestrating this failure type.

Notes

If no strategy is set for a failure type, the default strategy will be used when handling failures of that type.

get_handler(failure_type: str) → FailureHandler | None#

Get handler for a specific failure type.

Parameters:: failure_type (str) – Type identifier for the failure.
Returns:: Registered handler for the failure type, or None if no handler is registered.
Return type:: Optional[FailureHandler]

get_strategy(failure_type: str) → OrchestrationStrategy#

Get orchestration strategy for a specific failure type.

Parameters:: failure_type (str) – Type identifier for the failure.
Returns:: Configured strategy for the failure type, or the default strategy if no specific strategy is configured.
Return type:: OrchestrationStrategy

handle_failure(context: FailureContext) → Dict[str, Any]#

Handle a failure using the appropriate handler and strategy.

Parameters:

context (FailureContext) – Failure context containing all relevant information.

Returns:

Result of the handling operation.

Return type:

Dict[str, Any]

Raises:

ValueError – If the context is invalid or no handler is registered for the failure type.
RuntimeError – If handling fails due to handler errors or system state.

Notes

This method selects the appropriate handler based on the failure_type in the context, applies the configured strategy, and returns the handler’s result.

get_registered_types() → List[str]#

Get list of registered failure types.

Returns:: List of failure types that have handlers registered.
Return type:: List[str]

has_handler(failure_type: str) → bool#

Check if a handler is registered for a failure type.

Parameters:: failure_type (str) – Type identifier for the failure.
Returns:: True if a handler is registered for the failure type.
Return type:: bool

class qpanda3_runtime.toolbox.failure_orchestrator.FailureHandler#

Bases: abc.ABC

Abstract base class for failure handlers.

This class defines the interface for failure handlers that can process specific types of failures.

Subclasses should implement specific handling logic for different failure types and severity levels.

Examples

>>> class TimeoutHandler(FailureHandler):
...     def can_handle(self, context: FailureContext) -> bool:
...         return context.failure_type == "timeout"
...
...     def handle(self, context: FailureContext) -> Dict[str, Any]:
...         # Implement timeout handling logic
...         return {"action": "retry", "delay_ms": 1000}

abstractmethod can_handle(context: FailureContext) → bool#

Check if this handler can process the given failure context.

Parameters:: context (FailureContext) – Failure context to evaluate.
Returns:: True if this handler can process the failure, False otherwise.
Return type:: bool

Notes

Handlers should examine the failure_type, severity, and other context attributes to determine if they can handle the failure.

abstractmethod handle(context: FailureContext) → Dict[str, Any]#

Handle a failure based on the provided context.

Parameters:

context (FailureContext) – Failure context containing all relevant information.

Returns:

Result of the handling operation, including any actions taken or decisions made.

Return type:

Dict[str, Any]

Raises:

ValueError – If the context is invalid or missing required information.
RuntimeError – If handling fails due to external dependencies or system state.

class qpanda3_runtime.toolbox.failure_orchestrator.FailureContext#

Context information for a failure event.

Variables:

failure_type (str) – Type identifier for the failure.
severity (FailureSeverity) – Severity level of the failure.
timestamp (datetime) – When the failure occurred.
source (str) – Source component where failure originated.
data (Dict[str, Any]) – Additional failure-specific data.
trace_id (Optional[str]) – Associated trace ID for correlation with tracking system.

Examples

>>> context = FailureContext(
...     failure_type="timeout",
...     severity=FailureSeverity.MEDIUM,
...     timestamp=datetime.now(),
...     source="network_client",
...     data={"timeout_ms": 5000}
... )

failure_type: str#

severity: FailureSeverity#

timestamp: datetime.datetime#

source: str#

data: Dict[str, Any]#

trace_id: str | None = None#

class qpanda3_runtime.toolbox.failure_orchestrator.FailureSeverity(*args, **kwds)#

Bases: enum.Enum

Severity levels for failure classification.

Variables:

LOW (int) – Minor failure that doesn’t affect core functionality.
MEDIUM (int) – Moderate failure that affects some functionality.
HIGH (int) – Serious failure that affects core functionality.
CRITICAL (int) – Critical failure that requires immediate attention.

Examples

>>> severity = FailureSeverity.HIGH
>>> if severity.value >= FailureSeverity.HIGH.value:
...     send_alert()

LOW#

MEDIUM#

HIGH#

CRITICAL#

class qpanda3_runtime.toolbox.failure_orchestrator.OrchestrationStrategy(*args, **kwds)#

Bases: enum.Enum

Strategies for orchestrating failure handling.

Variables:

IMMEDIATE (int) – Handle failure immediately when detected.
DEFERRED (int) – Queue failure for later handling.
BATCH (int) – Batch multiple failures for efficient handling.
INTELLIGENT (int) – Use intelligent routing based on failure characteristics.

Notes

The strategy choice affects performance, resource usage, and failure recovery time.

IMMEDIATE#

DEFERRED#

BATCH#

INTELLIGENT#