python中的exec

u4vypkhs  于 5个月前  发布在  Python
关注(0)|答案(9)|浏览(57)

我想在一个exec函数中调用exec,并执行类似以下代码的操作(这是无效的):

import asyncio

async def f():
    await exec('x = 1\n' 'await asyncio.sleep(x)')

字符串
更准确地说,我希望能够在exec中运行的代码中等待未来。
如何才能做到这一点?

lpwwtiir

lpwwtiir1#

**注意:**仅python 3.6+支持F字符串。对于旧版本,请使用%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']()

字符串

已知问题:

  • 如果你在字符串中使用新行(三重引号),它会弄乱格式。
qyuhtwio

qyuhtwio2#

你的问题是,你试图等待None对象-exec忽略其代码的返回值,并总是返回None。如果你想执行并等待结果,你应该使用eval-eval返回给定表达式的值。
你的代码应该看起来像这样:

import asyncio

async def f():
    exec('x = 1')
    await eval('asyncio.sleep(x)')

loop = asyncio.get_event_loop()
loop.run_until_complete(f())
loop.close()

字符串

w8ntj3qf

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

字符串
它从保存局部变量开始。它声明了函数,但是使用了一个受限制的局部命名空间,因此它不会触及aexec帮助程序中声明的内容。函数名为func,我们访问locs dict,其中包含exec的局部变量的结果。locs["func"]是我们想要执行的,所以我们在aexec调用中使用**kwargs调用它,它将这些参数移动到本地命名空间。然后我们等待它并将其存储为result。最后,我们恢复本地变量并返回结果。

警告:

如果有任何多线程代码涉及全局变量,请不要使用此方法。使用@YouTwitFace的答案,它更简单且线程安全,或者删除全局保存/恢复代码

qco9c6ql

qco9c6ql4#

感谢所有的建议。我发现这可以用greenlets沿着along callc来完成,因为greenlets允许执行“顶级等待”:

import greenlet
import asyncio

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.value = None
            self.child.switch()

def gexec(code):
    child = greenlet.greenlet(exec)
    gawait = GreenAwait(child)
    child.switch(code, {'gawait': gawait})
    yield from gawait

async def aexec(code):
    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 = asyncio.get_event_loop()
    loop.run_until_complete(aexec(CODE + 'gawait(display_date())'))
    loop.close()

字符串

tnkciper

tnkciper5#

下面是使用内置ast模块的一种更健壮的方法:

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)

字符串

ctehm74n

ctehm74n6#

这里有一个module使用AST来做一些事情。这意味着多行字符串将完美地工作,行号将匹配原始语句。此外,如果任何东西是表达式,它将被返回(如果有多个,则作为列表返回,否则作为元素返回)
我做了这个模块(查看这个答案的修订历史以了解更多关于内部工作的细节)。我使用它here

cfh9epnr

cfh9epnr7#

使用这个函数:

import asyncio
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 asyncio
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]

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来获取返回值。

fivyi3re

fivyi3re8#

从Python 3.8开始,你可以compile()带有标志ast.PyCF_ALLOW_TOP_LEVEL_AWAIT的代码,eval它得到一个协程,然后你可以await。尽管使用eval,支持多个语句。
下面是如何做到这一点的最小示例:

import ast
import asyncio

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

v64noz0r

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)

字符串
然后,我可以运行:

>>> 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”命令要好。

相关问题