我有一个代码可以尝试代码块，如果发生错误，请重试直到最大计数。
该结构被多次重复使用。
看起来像这样：
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')
有没有办法创建for: try: ... except:raise的包装/装饰器/上下文管理器等，以便我可以重用该结构？类似于内联块或其他语言的匿名函数？
我无法创建函数，因为try:块可以包含任何语句，并使用任何变量。理想情况下，结构应该能够采用任意代码块。
例：
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)
编辑
我不想使用函数来包装语句的原因
- 我很懒，不想每次重用工作流时都为一次性使用创建一个函数。
- 该函数将有自己的作用域，这将使在我的代码中访问变量变得不简单
下面是上下文管理器的一个想法：
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=}')
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
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
完整代码(在线试用！
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=}')
相反，您可能会发现将所有内容打包到另一个函数中就是您所追求的
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})
目前尚不清楚为什么需要将任意代码语句"注入"try块中。如果您有由任意语句组成的过程，请将这些过程包装在函数定义中，并修饰函数：
import time
import logging
def retry(*, max_retries: int, sleep_interval: int):
def retry_wrapper(f):
def wrapped(*args, **kwargs):
for i in range(max_retries):
try:
logging.info(f"Trying function {f.__name__}")
return f(*args, **kwargs)
except Exception as error:
logging.info(f"retry {i}")
logging.error(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
如果您信任这些任意代码语句的来源，则可以将它们作为字符串传递并使用eval()但同样，我无法想象将过程包装到函数中并不更合适的场景。
以下是您可以使用与上述相同的装饰器执行此操作的一种方法：
@retry(max_retries=3, sleep_interval=1)
def arbitrary_code_runner(*statements):
for statement in statements:
eval(statement)
输出：
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
此方法的问题在于无法保存每个语句的结果。如果你的代码语句是突变器，那么这不是问题，但是如果你的任何任意语句依赖于其他语句的结果，你必须嵌套函数调用(就像我如何使用print(sum(...一样。
另一种技术是使用匿名和/或命名函数来存储任意语句，然后一次运行一个语句：
@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)}")
然后你可以用任意数量的 3 元组(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}))
输出：
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