我想在一个exec函数中调用exec,并执行类似以下代码的操作(这是无效的):
import asyncioasync def f(): await exec('x = 1\n' 'await asyncio.sleep(x)')
import asyncio
async def f():
await exec('x = 1\n' 'await asyncio.sleep(x)')
字符串更准确地说,我希望能够在exec中运行的代码中等待未来。如何才能做到这一点?
lpwwtiir1#
**注意:**仅python 3.6+支持F字符串。对于旧版本,请使用%s,.format()或经典的+串联。
%s
.format()
+
async def aexec(code): # Make an async function with the code and `exec` it exec( f'async def __ex(): ' + ''.join(f'\n {l}' for l in code.split('\n')) ) # Get `__ex` from local variables, call it and return the result return await locals()['__ex']()
async def aexec(code):
# Make an async function with the code and `exec` it
exec(
f'async def __ex(): ' +
''.join(f'\n {l}' for l in code.split('\n'))
)
# Get `__ex` from local variables, call it and return the result
return await locals()['__ex']()
字符串
qyuhtwio2#
你的问题是,你试图等待None对象-exec忽略其代码的返回值,并总是返回None。如果你想执行并等待结果,你应该使用eval-eval返回给定表达式的值。你的代码应该看起来像这样:
None
exec
eval
import asyncioasync def f(): exec('x = 1') await eval('asyncio.sleep(x)')loop = asyncio.get_event_loop()loop.run_until_complete(f())loop.close()
exec('x = 1')
await eval('asyncio.sleep(x)')
loop = asyncio.get_event_loop()
loop.run_until_complete(f())
loop.close()
w8ntj3qf3#
这是基于@YouTwitFace的答案,但保持全局变量不变,更好地处理局部变量并传递kwargs。注意多行字符串仍然不会保持其格式。也许你想要这个?
async def aexec(code, **kwargs): # Don't clutter locals locs = {} # Restore globals later globs = globals().copy() args = ", ".join(list(kwargs.keys())) exec(f"async def func({args}):\n " + code.replace("\n", "\n "), {}, locs) # Don't expect it to return from the coro. result = await locs["func"](**kwargs) try: globals().clear() # Inconsistent state finally: globals().update(**globs) return result
async def aexec(code, **kwargs):
# Don't clutter locals
locs = {}
# Restore globals later
globs = globals().copy()
args = ", ".join(list(kwargs.keys()))
exec(f"async def func({args}):\n " + code.replace("\n", "\n "), {}, locs)
# Don't expect it to return from the coro.
result = await locs["func"](**kwargs)
try:
globals().clear()
# Inconsistent state
finally:
globals().update(**globs)
return result
字符串它从保存局部变量开始。它声明了函数,但是使用了一个受限制的局部命名空间,因此它不会触及aexec帮助程序中声明的内容。函数名为func,我们访问locs dict,其中包含exec的局部变量的结果。locs["func"]是我们想要执行的,所以我们在aexec调用中使用**kwargs调用它,它将这些参数移动到本地命名空间。然后我们等待它并将其存储为result。最后,我们恢复本地变量并返回结果。
func
locs
locs["func"]
**kwargs
result
警告:
如果有任何多线程代码涉及全局变量,请不要使用此方法。使用@YouTwitFace的答案,它更简单且线程安全,或者删除全局保存/恢复代码
qco9c6ql4#
感谢所有的建议。我发现这可以用greenlets沿着along callc来完成,因为greenlets允许执行“顶级等待”:
import greenletimport asyncioclass GreenAwait: def __init__(self, child): self.current = greenlet.getcurrent() self.value = None self.child = child def __call__(self, future): self.value = future self.current.switch() def __iter__(self): while self.value is not None: yield self.value self.value = None self.child.switch()def gexec(code): child = greenlet.greenlet(exec) gawait = GreenAwait(child) child.switch(code, {'gawait': gawait}) yield from gawaitasync def aexec(code): green = greenlet.greenlet(gexec) gen = green.switch(code) for future in gen: await future# modified asyncio example from Python docsCODE = ('import asyncio\n' 'import datetime\n' 'async def display_date():\n' ' for i in range(5):\n' ' print(datetime.datetime.now())\n' ' await asyncio.sleep(1)\n')def loop(): loop = asyncio.get_event_loop() loop.run_until_complete(aexec(CODE + 'gawait(display_date())')) loop.close()
import greenlet
class GreenAwait:
def __init__(self, child):
self.current = greenlet.getcurrent()
self.value = None
self.child = child
def __call__(self, future):
self.value = future
self.current.switch()
def __iter__(self):
while self.value is not None:
yield self.value
self.child.switch()
def gexec(code):
child = greenlet.greenlet(exec)
gawait = GreenAwait(child)
child.switch(code, {'gawait': gawait})
yield from gawait
green = greenlet.greenlet(gexec)
gen = green.switch(code)
for future in gen:
await future
# modified asyncio example from Python docs
CODE = ('import asyncio\n'
'import datetime\n'
'async def display_date():\n'
' for i in range(5):\n'
' print(datetime.datetime.now())\n'
' await asyncio.sleep(1)\n')
def loop():
loop.run_until_complete(aexec(CODE + 'gawait(display_date())'))
tnkciper5#
下面是使用内置ast模块的一种更健壮的方法:
ast
import astasync def async_exec(stmts, env=None): parsed_stmts = ast.parse(stmts) fn_name = "_async_exec_f" fn = f"async def {fn_name}(): pass" parsed_fn = ast.parse(fn) for node in parsed_stmts.body: ast.increment_lineno(node) parsed_fn.body[0].body = parsed_stmts.body exec(compile(parsed_fn, filename="<ast>", mode="exec"), env) return await eval(f"{fn_name}()", env)
import ast
async def async_exec(stmts, env=None):
parsed_stmts = ast.parse(stmts)
fn_name = "_async_exec_f"
fn = f"async def {fn_name}(): pass"
parsed_fn = ast.parse(fn)
for node in parsed_stmts.body:
ast.increment_lineno(node)
parsed_fn.body[0].body = parsed_stmts.body
exec(compile(parsed_fn, filename="<ast>", mode="exec"), env)
return await eval(f"{fn_name}()", env)
ctehm74n6#
这里有一个module使用AST来做一些事情。这意味着多行字符串将完美地工作,行号将匹配原始语句。此外,如果任何东西是表达式,它将被返回(如果有多个,则作为列表返回,否则作为元素返回)我做了这个模块(查看这个答案的修订历史以了解更多关于内部工作的细节)。我使用它here
cfh9epnr7#
使用这个函数:
import asyncioasync def async_exec(code): t = [None] exec('async def _async_exec():\n return {}\nt[0] = asyncio.ensure_future(_async_exec())'.format(code)) return await t[0]
async def async_exec(code):
t = [None]
exec('async def _async_exec():\n return {}\nt[0] = asyncio.ensure_future(_async_exec())'.format(code))
return await t[0]
字符串下面是一个可以直接运行的代码示例。(适用于Python 3.6.8)
import asyncioasync def async_exec(code): t = [None] exec('async def _async_exec():\n return {}\nt[0] = asyncio.ensure_future(_async_exec())'.format(code)) return await t[0]async def p(s): await asyncio.sleep(s) return sasync def main(): print(await async_exec('await p(0.1) / await p(0.2)'))asyncio.get_event_loop().run_until_complete(main())
async def p(s):
await asyncio.sleep(s)
return s
async def main():
print(await async_exec('await p(0.1) / await p(0.2)'))
asyncio.get_event_loop().run_until_complete(main())
型我试着解释一下,在exec中定义一个future函数。在future函数中,运行你想要的代码。但是exec没有返回值,使用t[0]存储一个future,等待exec外部的future来获取返回值。
fivyi3re8#
从Python 3.8开始,你可以compile()带有标志ast.PyCF_ALLOW_TOP_LEVEL_AWAIT的代码,eval它得到一个协程,然后你可以await。尽管使用eval,支持多个语句。下面是如何做到这一点的最小示例:
compile()
ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
await
import astimport asyncioasync def main() -> None: code = compile( "x = 1\n" "await asyncio.sleep(0.1)\n" "print('done!')\n", "<string>", "exec", flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT, ) coroutine: Awaitable | None = eval(code) if coroutine is not None: await coroutineasyncio.run(main())
async def main() -> None:
code = compile(
"x = 1\n"
"await asyncio.sleep(0.1)\n"
"print('done!')\n",
"<string>",
"exec",
flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT,
coroutine: Awaitable | None = eval(code)
if coroutine is not None:
await coroutine
asyncio.run(main())
字符串当没有await语句时,eval()立即运行代码,然后返回None。
eval()
v64noz0r9#
根据不同的答案,我唯一遗漏的是局部变量(不能使它们成为全局变量)。我是这么做的:
async def async_eval_or_exec(message, vars): # Translating the command into a function # The function returns the main result AND the local variables to make them accessible outside if message.count("\n"): function = "async def __execute__():\n " + message.replace("\n", " ") + "\n return None, locals()\n" else: function = "async def __execute__():\n return " + message + ", locals()\n" # The execution - get the result of try: exec(function, vars, vars) result = await vars["__execute__"] () if result[ 0 ] is not None: return result[ 0 ] vars.update(result[ 1 ]) # forces the local variables (inside __execute__) to be global except SyntaxError: # for commands like "import os" exec(message, vars, vars)
async def async_eval_or_exec(message, vars):
# Translating the command into a function
# The function returns the main result AND the local variables to make them accessible outside
if message.count("\n"):
function = "async def __execute__():\n " + message.replace("\n", " ") + "\n return None, locals()\n"
else:
function = "async def __execute__():\n return " + message + ", locals()\n"
# The execution - get the result of
exec(function, vars, vars)
result = await vars["__execute__"] ()
if result[ 0 ] is not None:
return result[ 0 ]
vars.update(result[ 1 ]) # forces the local variables (inside __execute__) to be global
except SyntaxError: # for commands like "import os"
exec(message, vars, vars)
字符串然后,我可以运行:
>>> vars = {}>>> await async_eval_or_exec('x = 1', vars)>>> await async_eval_or_exec('await asyncio.sleep(x)', vars)>>> await async_eval_or_exec('print(x)', vars)1>>>
>>> vars = {}
>>> await async_eval_or_exec('x = 1', vars)
>>> await async_eval_or_exec('await asyncio.sleep(x)', vars)
>>> await async_eval_or_exec('print(x)', vars)
1
>>>
型如果你创建了一个JavaScript解释器,它会很有用(为了不丢失你在execute函数中创建的对象)。我认为这比每次创建变量时都使用“global”命令要好。
9条答案
按热度按时间lpwwtiir1#
**注意:**仅python 3.6+支持F字符串。对于旧版本,请使用
%s
,.format()
或经典的+
串联。字符串
已知问题:
qyuhtwio2#
你的问题是,你试图等待
None
对象-exec
忽略其代码的返回值,并总是返回None
。如果你想执行并等待结果,你应该使用eval
-eval
返回给定表达式的值。你的代码应该看起来像这样:
字符串
w8ntj3qf3#
这是基于@YouTwitFace的答案,但保持全局变量不变,更好地处理局部变量并传递kwargs。注意多行字符串仍然不会保持其格式。也许你想要这个?
字符串
它从保存局部变量开始。它声明了函数,但是使用了一个受限制的局部命名空间,因此它不会触及aexec帮助程序中声明的内容。函数名为
func
,我们访问locs
dict,其中包含exec的局部变量的结果。locs["func"]
是我们想要执行的,所以我们在aexec调用中使用**kwargs
调用它,它将这些参数移动到本地命名空间。然后我们等待它并将其存储为result
。最后,我们恢复本地变量并返回结果。警告:
如果有任何多线程代码涉及全局变量,请不要使用此方法。使用@YouTwitFace的答案,它更简单且线程安全,或者删除全局保存/恢复代码
qco9c6ql4#
感谢所有的建议。我发现这可以用greenlets沿着along callc来完成,因为greenlets允许执行“顶级等待”:
字符串
tnkciper5#
下面是使用内置
ast
模块的一种更健壮的方法:字符串
ctehm74n6#
这里有一个module使用AST来做一些事情。这意味着多行字符串将完美地工作,行号将匹配原始语句。此外,如果任何东西是表达式,它将被返回(如果有多个,则作为列表返回,否则作为元素返回)
我做了这个模块(查看这个答案的修订历史以了解更多关于内部工作的细节)。我使用它here
cfh9epnr7#
使用这个函数:
字符串
下面是一个可以直接运行的代码示例。(适用于Python 3.6.8)
型
我试着解释一下,在exec中定义一个future函数。在future函数中,运行你想要的代码。但是exec没有返回值,使用t[0]存储一个future,等待exec外部的future来获取返回值。
fivyi3re8#
从Python 3.8开始,你可以
compile()
带有标志ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
的代码,eval
它得到一个协程,然后你可以await
。尽管使用eval
,支持多个语句。下面是如何做到这一点的最小示例:
字符串
当没有
await
语句时,eval()
立即运行代码,然后返回None
。v64noz0r9#
根据不同的答案,我唯一遗漏的是局部变量(不能使它们成为全局变量)。
我是这么做的:
字符串
然后,我可以运行:
型
如果你创建了一个JavaScript解释器,它会很有用(为了不丢失你在execute函数中创建的对象)。
我认为这比每次创建变量时都使用“global”命令要好。