I have a code that tries a block of code, if error happens, retry up to a max count.
The structure is re-used many times.
Something looks like this:
a = 10
b = 20
c = 30
result = None
max_loop = 3
interval = 1
for i in range(max_loop):
try:
# these are generic statements
statement_1(a)
statement_2(b)
# ...
result = statement_3(c)
break
except Exception as e:
logging.error(f'retry {i}')
logging.error(e)
time.sleep(interval)
raise Exception(f'Failed after {max_loop} retries')
Is there a way to create a wrap/decorator/contextmanager, etc of the for: try: ... except:raise so I can reuse the structure? Something similar to in-line block, or anonymous function in other languages?
I cannot create a function because the try: block can contain any statements, and use any variables. Ideally, the structure should be able to take arbitrary code block.
example:
repeat_try_and_raise:
statement_1(a)
statement_2(b)
# ...
result = statement_3(c)
# ...
repeat_try_and_raise:
statement_4(a)
statement_5(b)
# ...
statement_6(c)
CodePudding user response：
Instead, you may find just packing all of 'em into another function is what you're after
MAX_ATTEMPTS = 3
def function_collection(arg1, arg2, arg3):
fn1(arg1)
fn2(arg2)
fn3(arg3)
def wrapper(target, target_kwargs, max_attempts=MAX_ATTEMPTS):
assert max_attempts >= 2
for attempt in range(max_attempts):
try:
return target(**fn_kwargs)
except SubClassException:
continue
except Exception as ex:
LOGGER.warning(f"something unexpected went wrong: {repr(ex)}")
raise OSError(f"no run of target succeeded after {max_attempts} attempts")
wrapper(function_collection, {"arg1": 10, "arg2": 20, "arg3": 30})
CodePudding user response：
It's not clear why you need to "inject" arbitrary code statements into the try block. If you have procedures that are made up of arbitrary statements, wrap the procedures up in a function definition, and decorate the function:
def retry(*, max_retries: int, sleep_interval: int):
def retry_wrapper(f):
def wrapped(*args, **kwargs):
for i in range(max_retries):
try:
print(f"Trying function {f.__name__}")
return f(*args, **kwargs)
except Exception as error:
print(f"retry {i}: {error}")
time.sleep(sleep_interval)
raise Exception(f"Failed after {max_retries} tries")
return wrapped
return retry_wrapper
@retry(max_retries=6, sleep_interval=2)
def procedure_1(*args, **kwargs):
# statement 1
# statement 2
# statement 3
pass
@retry(max_retries=3, sleep_interval=1)
def procedure_2(*args, **kwargs):
# statement 4
# statement 5
# statement 6
pass
If you trust the source of these arbitrary code statements, you could pass them as strings and use eval() but again, I cannot imagine a scenario where wrapping your procedures into a function isn't more appropriate.
Here's one way you might do this, using the same decorator as above:
@retry(max_retries=3, sleep_interval=1)
def arbitrary_code_runner(*statements):
for statement in statements:
eval(statement)
Output:
In [5]: arbitrary_code_runner("print('hello world')", "print(sum(x**2 for x in range(10)))", "print('I am arbitrary code, I am potentially dangerous')")
Trying function arbitrary_code_runner
hello world
285
I am arbitrary code, I am potentially dangerous
The problem with this approach is that you cannot save the results of each statement. If your code statements are mutators then this is not a problem, but if any of your arbitrary statements rely on the results of other statements, you'll have to nest the function calls (like how I used print(sum(....
Another technique would be to use anonymous and/or named functions to store your arbitrary statements, and then run each statement one at a time:
@retry(max_retries=3, sleep_interval=1)
def arbitrary_function_runner(*funcs_and_args_and_kwargs):
for func, args, kwargs in funcs_and_args_and_kwargs:
print(f"\n Function {func.__name__} called")
print(f" args: {', '.join(map(str, args))}")
print(f" kwargs: {', '.join(f'{key}: {value}' for key, value in kwargs.items())}")
print(f" result: {func(*args, **kwargs)}")
Which you could then call with an arbitrary number of 3-tuples of (function, args tuple, kwargs dict):
def some_named_function(*args, **kwargs):
return "some named function's results"
arbitrary_function_runner((lambda *args, **kwargs: "".join(kwargs[arg] for arg in args), ("a", "b", "c"), {"a": "A", "b": "B", "c": "C"}),
(lambda x: x**2, (3,), {}),
(some_named_function, (1, 2, 3), {"kwarg1": 1, "kwarg_2": 2}))
Output:
Trying function arbitrary_function_runner
Function <lambda> called
args: a, b, c
kwargs: a: A, b: B, c: C
result: ABC
Function <lambda> called
args: 3
kwargs:
result: 9
Function some_named_function called
args: 1, 2, 3
kwargs: kwarg1: 1, kwarg_2: 2
result: some named function's results
CodePudding user response：
Here's an idea with a context manager:
a = 10
result = None
max_loop = 2
interval = 0.1
for attempt in repeat_try_and_raise(max_loop, interval):
with attempt:
a = 1
result = 0/0 if a < 13 else 42
print(f'Success with {result=}')
Output for max_loop = 2:
ERROR:root:retry 0
ERROR:root:division by zero
ERROR:root:retry 1
ERROR:root:division by zero
Traceback (most recent call last):
File ".code.tio", line 25, in <module>
for attempt in repeat_try_and_raise(max_loop, interval):
File ".code.tio", line 18, in repeat_try_and_raise
raise Exception(f'Failed after {max_loop} retries')
Exception: Failed after 2 retries
Output for max_loop = 3:
ERROR:root:retry 0
ERROR:root:division by zero
ERROR:root:retry 1
ERROR:root:division by zero
Success with result=42
Full code (Try it online!):
import logging, time
def repeat_try_and_raise(max_loop, interval):
class Attempt:
def __enter__(self):
pass
def __exit__(self, exc_type, exc_value, traceback):
self.e = exc_value
return True
for i in range(max_loop):
attempt = Attempt()
yield attempt
if attempt.e is None:
return
logging.error(f'retry {i}')
logging.error(attempt.e)
time.sleep(interval)
raise Exception(f'Failed after {max_loop} retries')
a = 10
result = None
max_loop = 3
interval = 0.1
for attempt in repeat_try_and_raise(max_loop, interval):
with attempt:
a = 1
result = 0/0 if a < 13 else 42
print(f'Success with {result=}')