Coverage for src/zapy/templating/eval.py: 100%
35 statements
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
« prev ^ index » next coverage.py v7.4.1, created at 2024-02-10 19:35 +0000
1import ast
2from typing import Any, cast
4from .traceback import annotate_traceback
6_python_code = """
7async def __exec_wrapper(_globs):
8 globals().update(**_globs)
9 __placeholder
10 return locals()
11"""
14async def exec_async(custom_code: str, _globals: dict) -> None:
15 parsed_ast = ast.parse(_python_code)
17 def is_placeholder(node: ast.AST) -> bool:
18 return isinstance(node, ast.Expr) and isinstance(node.value, ast.Name) and node.value.id == "__placeholder"
20 for node in ast.walk(parsed_ast):
21 if is_placeholder(node):
22 custom_ast = ast.parse(custom_code)
23 node = cast(ast.Expr, node)
24 new_node = cast(ast.expr, custom_ast)
25 node.value = new_node
27 unparsed = ast.unparse(parsed_ast)
28 exec(unparsed) # noqa S102
30 try:
31 func = locals()["__exec_wrapper"]
32 new_locals = await func(_globals)
33 except BaseException as e:
34 annotate_traceback(e, unparsed, location="exec_async")
35 raise
37 _globals.update(new_locals)
40def exec_sync(custom_code: str, _globals: dict) -> None:
41 try:
42 return exec(custom_code, _globals) # noqa S102
43 except BaseException as e:
44 annotate_traceback(e, custom_code, location="exec_sync")
45 raise
48def eval_sync(custom_code: str, _globals: dict) -> Any:
49 try:
50 return eval(custom_code, _globals) # noqa S307
51 except BaseException as e:
52 annotate_traceback(e, custom_code, location="eval_sync")
53 raise