WireRope Library

Warnings

• gotcha When using `timeout`, the underlying function or operation might continue running in the background even after `TimeoutError` is raised. `WireRope` interrupts the call from the caller's perspective, but does not inherently cancel or terminate the wrapped function's execution, especially for I/O-bound or native code. This can lead to resource leaks or unexpected side effects.
  Fix: Be aware of the wrapped function's behavior during a timeout. If true cancellation is required, the wrapped function itself must be designed to be cancelable (e.g., by checking a flag, using `asyncio.CancelledError`, or implementing specific resource cleanup).
• gotcha Employing the `retries` feature without ensuring the wrapped function is idempotent or handles multiple executions gracefully can lead to unintended side effects. Each retry re-executes the function, potentially causing duplicate actions if the function isn't designed for it.
  Fix: Design the wrapped function to be idempotent (producing the same result and side effects regardless of how many times it's executed) or to explicitly handle retry attempts (e.g., by checking for existing data before performing an action).
• gotcha When wrapping class methods or functions that rely on specific contextual state, `WireRope` creates a new execution context. If the wrapped callable needs to access or modify external state (e.g., `self` attributes in a method), ensure `WireRope` is initialized with the correct `self` reference or that state is passed explicitly. Direct modification of `self` might not be reflected as expected without careful handling.
  Fix: For class methods, ensure `WireRope` is instantiated correctly within the class context or passed the `self` instance. For complex state management, consider passing necessary state explicitly as arguments to the wrapped function or using `wirerope.context` primitives if your use case aligns with its design for context propagation.

Install

• pip install wirerope Install stable version

Imports

• WireRope

  from wirerope import WireRope

Quickstart

This quickstart demonstrates how to wrap a basic Python function with `WireRope` to gain control over its execution. It then shows more advanced usage with timeout functionality, where `WireRope` raises a `TimeoutError` if the wrapped function exceeds a specified duration, and how to configure automatic retries for flaky operations.

import time
from wirerope import WireRope

def my_function(x, y):
    print(f"Executing my_function with {x}, {y}")
    return x + y

# Wrap a regular function
rope_func = WireRope(my_function)
result_func = rope_func.call(10, 20)
print(f"Result from wrapped function: {result_func}")

# Example with timeout control
def long_task():
    print("Long task started...")
    time.sleep(2) # Simulate work
    print("Long task finished.")
    return "Task Complete"

# Set a timeout shorter than the task duration
rope_timeout = WireRope(long_task, timeout=1)
try:
    timeout_result = rope_timeout.call()
    print(f"Timeout Result: {timeout_result}")
except TimeoutError:
    print("Task timed out as expected after 1 second.")

# Example with retries
def flaky_task(attempt):
    if attempt < 2:
        raise ValueError("Simulating a temporary failure")
    return "Success after retries"

# WireRope automatically passes an 'attempt' argument to the wrapped function
# when 'retries' is enabled. The first call is attempt=0.
rope_retries = WireRope(flaky_task, retries=3)
try:
    # The 'attempt' argument will be automatically injected by WireRope
    # Note: flaky_task would typically infer the attempt count internally or not use it directly
    # For this example, we demonstrate it expecting an 'attempt' parameter
    # In real world, your function might just fail and WireRope retries it
    result_retries = rope_retries.call(attempt=0)
    print(f"Result from retried task: {result_retries}")
except Exception as e:
    print(f"Failed after retries: {e}")