Plugins Module¶

The plugins module provides built-in plugin implementations for extending cascade stage handlers with caching, retry logic, metrics collection, and circuit breaking.

CachePlugin¶

Bases: StagePlugin

Plugin for caching stage execution results.

Caches results based on input context hash with configurable TTL. Cache entries expire after ttl_seconds.

Attributes:

Name	Type	Description
`wrapped_handler`		The underlying handler to cache
`ttl_seconds`		Time-to-live for cache entries in seconds

`name: str` `property` ¶

Get plugin name.

`init(wrapped_handler: Callable, ttl_seconds: int = 300)` ¶

Initialize cache plugin.

Parameters:

Name	Type	Description	Default
`wrapped_handler`	`Callable`	Handler function to wrap with caching	required
`ttl_seconds`	`int`	Cache entry TTL in seconds (default: 300)	`300`

`execute(context: Dict[str, Any]) -> Dict[str, Any]` `async` ¶

Execute with caching.

Checks cache for existing result. If cache miss or expired, executes wrapped handler and stores result.

Parameters:

Name	Type	Description	Default
`context`	`Dict[str, Any]`	Execution context	required

Returns:

Type	Description
`Dict[str, Any]`	Cached or newly computed result

Raises:

Type	Description
`Exception`	Any errors from wrapped handler

RetryPlugin¶

Bases: StagePlugin

Plugin for retry logic with exponential backoff.

Retries failed executions with exponential backoff between attempts.

Attributes:

Name	Type	Description
`wrapped_handler`		The underlying handler to retry
`max_retries`		Maximum number of retry attempts
`delay_ms`		Initial delay between retries in milliseconds

`name: str` `property` ¶

Get plugin name.

`init(wrapped_handler: Callable, max_retries: int = 3, delay_ms: int = 100)` ¶

Initialize retry plugin.

Parameters:

Name	Type	Description	Default
`wrapped_handler`	`Callable`	Handler function to wrap with retries	required
`max_retries`	`int`	Maximum retry attempts (default: 3)	`3`
`delay_ms`	`int`	Initial delay in milliseconds (default: 100)	`100`

`execute(context: Dict[str, Any]) -> Dict[str, Any]` `async` ¶

Execute with retry logic.

Attempts execution up to max_retries times with exponential backoff. Delay doubles after each failed attempt.

Parameters:

Name	Type	Description	Default
`context`	`Dict[str, Any]`	Execution context	required

Returns:

Type	Description
`Dict[str, Any]`	Result from successful execution

Raises:

Type	Description
`Exception`	Last exception if all retries exhausted

MetricsPlugin¶

Bases: StagePlugin

Plugin for collecting execution metrics.

Tracks execution count, total time, errors, and computes success rate and average execution time.

Attributes:

Name	Type	Description
`wrapped_handler`		The underlying handler to monitor
`count`		Total number of executions
`total_time_ms`		Total execution time in milliseconds
`errors`		Number of failed executions

`name: str` `property` ¶

Get plugin name.

`success_rate: float` `property` ¶

Calculate success rate.

Returns:

Type	Description
`float`	Success rate as percentage (0-100)

`avg_time_ms: float` `property` ¶

Calculate average execution time.

Returns:

Type	Description
`float`	Average time in milliseconds

`metrics: Dict[str, Any]` `property` ¶

Get current metrics.

Returns:

Type	Description
`Dict[str, Any]`	Dictionary containing all metrics

`init(wrapped_handler: Callable)` ¶

Initialize metrics plugin.

Parameters:

Name	Type	Description	Default
`wrapped_handler`	`Callable`	Handler function to wrap with metrics	required

`execute(context: Dict[str, Any]) -> Dict[str, Any]` `async` ¶

Execute with metrics collection.

Times execution and updates metrics counters.

Parameters:

Name	Type	Description	Default
`context`	`Dict[str, Any]`	Execution context	required

Returns:

Type	Description
`Dict[str, Any]`	Result from wrapped handler

Raises:

Type	Description
`Exception`	Any errors from wrapped handler (recorded in metrics)

CircuitBreakerPlugin¶

Bases: StagePlugin

Plugin implementing circuit breaker pattern.

Prevents cascading failures by opening circuit after threshold failures. Circuit automatically resets after timeout period.

States

CLOSED: Normal operation, requests pass through
OPEN: Circuit tripped, requests fail immediately
HALF_OPEN: Testing if service recovered (auto after timeout)

Attributes:

Name	Type	Description
`wrapped_handler`		The underlying handler to protect
`failure_threshold`		Number of failures before opening circuit
`reset_timeout_seconds`		Time before attempting reset
`failure_count`		Current consecutive failure count
`last_failure_time`	`Optional[float]`	Timestamp of last failure
`is_open`		Whether circuit is currently open

`name: str` `property` ¶

Get plugin name.

`init(wrapped_handler: Callable, failure_threshold: int = 5, reset_timeout_seconds: int = 60)` ¶

Initialize circuit breaker plugin.

Parameters:

Name	Type	Description	Default
`wrapped_handler`	`Callable`	Handler function to protect	required
`failure_threshold`	`int`	Failures before opening (default: 5)	`5`
`reset_timeout_seconds`	`int`	Timeout before reset attempt (default: 60)	`60`

`execute(context: Dict[str, Any]) -> Dict[str, Any]` `async` ¶

Execute with circuit breaker protection.

Checks circuit state before execution. If open, fails immediately unless timeout has passed (half-open state).

Parameters:

Name	Type	Description	Default
`context`	`Dict[str, Any]`	Execution context	required

Returns:

Type	Description
`Dict[str, Any]`	Result from successful execution

Raises:

Type	Description
`RuntimeError`	If circuit is open
`Exception`	Any errors from wrapped handler

PluginFactory¶

Factory for composing plugins around handlers.

Provides utilities for wrapping handlers with multiple plugins in a composable manner. Plugins are applied in reverse order so that the first plugin in the list is the outermost wrapper.

`wrap_handler(handler: Callable, plugins: List[str], config: Dict[str, Any]) -> Callable` `async` `staticmethod` ¶

Wrap a handler with multiple plugins.

Applies plugins in reverse order to create a plugin chain: cache -> retry -> metrics -> circuit_breaker -> handler

Parameters:

Name	Type	Description	Default
`handler`	`Callable`	Base handler to wrap	required
`plugins`	`List[str]`	List of plugin names to apply (e.g., ["cache", "retry"])	required
`config`	`Dict[str, Any]`	Configuration dict with plugin-specific settings Example: { "cache": {"ttl_seconds": 600}, "retry": {"max_retries": 5}, "circuit_breaker": {"failure_threshold": 3} }	required

Returns:

Type	Description
`Callable`	Wrapped handler with all plugins applied

Raises:

Type	Description
`ValueError`	If unknown plugin name provided

Plugins Module¶

CachePlugin¶

name: str property ¶

__init__(wrapped_handler: Callable, ttl_seconds: int = 300) ¶

execute(context: Dict[str, Any]) -> Dict[str, Any] async ¶

RetryPlugin¶

name: str property ¶

__init__(wrapped_handler: Callable, max_retries: int = 3, delay_ms: int = 100) ¶

execute(context: Dict[str, Any]) -> Dict[str, Any] async ¶

MetricsPlugin¶

name: str property ¶

success_rate: float property ¶

avg_time_ms: float property ¶

metrics: Dict[str, Any] property ¶

__init__(wrapped_handler: Callable) ¶

execute(context: Dict[str, Any]) -> Dict[str, Any] async ¶

CircuitBreakerPlugin¶

name: str property ¶

__init__(wrapped_handler: Callable, failure_threshold: int = 5, reset_timeout_seconds: int = 60) ¶

execute(context: Dict[str, Any]) -> Dict[str, Any] async ¶

PluginFactory¶

wrap_handler(handler: Callable, plugins: List[str], config: Dict[str, Any]) -> Callable async staticmethod ¶

`name: str` `property` ¶

`init(wrapped_handler: Callable, ttl_seconds: int = 300)` ¶

`execute(context: Dict[str, Any]) -> Dict[str, Any]` `async` ¶

`name: str` `property` ¶

`init(wrapped_handler: Callable, max_retries: int = 3, delay_ms: int = 100)` ¶

`execute(context: Dict[str, Any]) -> Dict[str, Any]` `async` ¶

`name: str` `property` ¶

`success_rate: float` `property` ¶

`avg_time_ms: float` `property` ¶

`metrics: Dict[str, Any]` `property` ¶

`init(wrapped_handler: Callable)` ¶

`execute(context: Dict[str, Any]) -> Dict[str, Any]` `async` ¶

`name: str` `property` ¶

`init(wrapped_handler: Callable, failure_threshold: int = 5, reset_timeout_seconds: int = 60)` ¶

`execute(context: Dict[str, Any]) -> Dict[str, Any]` `async` ¶

`wrap_handler(handler: Callable, plugins: List[str], config: Dict[str, Any]) -> Callable` `async` `staticmethod` ¶