wirerope

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}")