使用 exec() 动态调用异步函数
Call Async Functions Dynamically with exec()
所以,我正在使用 discord.py 实现一个 Discord Bot,我正在尝试根据命令动态调用函数。我能够使用 exec() 测试动态函数调用,但它们似乎与 discord.py.
所需的异步调用分崩离析
因此,我在这个示例中尝试做的是调用 hello 函数,并通过在聊天中键入 !hello 将 Hello World 打印到 discord 中。
@client.event
async def on_message(message):
call = 'await ' + message.content.lower()[1:] + '(message)'
exec(call)
async def hello(message):
await client.send_message(message.channel, 'Hello World')
不幸的是,这段代码似乎没有做任何事情,我假设是因为 exec() 如何处理异步调用。任何帮助将不胜感激。
而不是 exec()
使用 globals()
来获取您的函数:
import asyncio
async def main():
s = "foo"
param = "hello"
coro = globals().get(s)
if coro:
result = await coro(param)
print("got:", result)
else:
print("unknown function:", s)
async def foo(param):
print(param)
await asyncio.sleep(0.11)
return ":-)"
loop = asyncio.get_event_loop()
response = loop.run_until_complete(main())
loop.close()
但是,允许用户访问 globals()
中的任何内容可能很危险,相反,将您的命令列入白名单会更好,例如使用:
import asyncio
my_commands = {}
def register(cmd):
my_commands[cmd.__name__] = cmd
return cmd
async def main():
s = "foo"
param = "hello"
coro = my_commands.get(s)
if coro:
result = await coro(param)
print("got:", result)
else:
print("unknown function:", s)
@register
async def foo(param):
"""I am the mighty foo command!"""
print(param)
await asyncio.sleep(0.11)
return ":-)"
loop = asyncio.get_event_loop()
response = loop.run_until_complete(main())
loop.close()
另请参阅:
for k, v in my_commands.items():
print("{}: {}".format(k, v.__doc__ or "no docs"))
所以,我正在使用 discord.py 实现一个 Discord Bot,我正在尝试根据命令动态调用函数。我能够使用 exec() 测试动态函数调用,但它们似乎与 discord.py.
所需的异步调用分崩离析因此,我在这个示例中尝试做的是调用 hello 函数,并通过在聊天中键入 !hello 将 Hello World 打印到 discord 中。
@client.event
async def on_message(message):
call = 'await ' + message.content.lower()[1:] + '(message)'
exec(call)
async def hello(message):
await client.send_message(message.channel, 'Hello World')
不幸的是,这段代码似乎没有做任何事情,我假设是因为 exec() 如何处理异步调用。任何帮助将不胜感激。
而不是 exec()
使用 globals()
来获取您的函数:
import asyncio
async def main():
s = "foo"
param = "hello"
coro = globals().get(s)
if coro:
result = await coro(param)
print("got:", result)
else:
print("unknown function:", s)
async def foo(param):
print(param)
await asyncio.sleep(0.11)
return ":-)"
loop = asyncio.get_event_loop()
response = loop.run_until_complete(main())
loop.close()
但是,允许用户访问 globals()
中的任何内容可能很危险,相反,将您的命令列入白名单会更好,例如使用:
import asyncio
my_commands = {}
def register(cmd):
my_commands[cmd.__name__] = cmd
return cmd
async def main():
s = "foo"
param = "hello"
coro = my_commands.get(s)
if coro:
result = await coro(param)
print("got:", result)
else:
print("unknown function:", s)
@register
async def foo(param):
"""I am the mighty foo command!"""
print(param)
await asyncio.sleep(0.11)
return ":-)"
loop = asyncio.get_event_loop()
response = loop.run_until_complete(main())
loop.close()
另请参阅:
for k, v in my_commands.items():
print("{}: {}".format(k, v.__doc__ or "no docs"))