如何从另一个线程中的同步代码调用异步代码？

Question

我正在制作一个 Discord 机器人，它会在收到 Github 钩子时发送 PM。

它使用 Discord.py 和 BottlePy，最后一个在专用线程中运行。 因为这两个框架都有一个阻塞的主循环。

在 BottlePy 回调中，我调用了一些 Discord.py 异步代码。

我不知道什么是 Python 异步，当与同步代码混合时，这似乎很复杂......

这是完整的源代码：

import discord
import bottle
import threading
import asyncio

client = discord.Client()
server = bottle.Bottle()

async def dm_on_github_async(userid,request):
    print("Fire Discord dm to "+str(userid))
    global client
    user = client.get_user(userid)
    if (user==None):
        abort(500, "User lookup failed");

    dm_channel = user.dm_channel
    if (dm_channel==None):
        dm_channel = await user.create_dm()
    if (dm_channel==None):
        abort(500, "Fail to create DM channel");
    print("DM channel is "+str(asyncio.wait(dm_channel.id)))
    await dm_channel.send("There's a Github shot !")
    await dm_channel.send(str(request.body))
    return
@server.post("/dm_on_github/<userid:int>")
def dm_on_github(userid):
    return asyncio.run(dm_on_github_async(userid,bottle.request))
@client.event
async def on_ready():
    print('We have logged in as {0.user} '.format(client))

#@client.event
#async def on_message(message):
#    if message.author == client.user:
#        return
#
#    if message.content.startswith('$hello'):
#        await message.channel.send('Hello!')
#    # This sample was working very well

class HTTPThread(threading.Thread):
    def run(self):
        global server
        server.run(port=8080)
server_thread = HTTPThread()
print("Starting HTTP server")
server_thread.start()
print("Starting Discord client")
client.run('super secret key')
print("Client terminated")
server.close()
print("Asked server to terminate")
server_thread.join()
print("Server thread successful join")

我希望我的 Python 机器人将 HTTP 请求的正文作为 PM 发送。

我在return asyncio.run(dm_on_github_async(userid,bottle.request)) 收到RuntimeError: Timeout context manager should be used inside a task。

我认为我没有以正确的方式进行这种混合......

Answer 1

一夜之后，我找到了路。

要从另一个线程中的同步代码调用异步代码，我们要求loop（这里是来自 Discord.py 的这个）使用asyncio.run_coroutine_threadsafe() 运行回调，这会返回一个Task()，我们等待他的结果result()。

回调将在循环线程中运行，在我的情况下，我需要 copy() Bottle 请求。

这是一个工作程序（只要您不介意停止它...）：

import discord
import bottle
import threading
import asyncio

client = discord.Client()
server = bottle.Bottle()
class HTTPThread(threading.Thread):
    def run(self):
        global server
        server.run(port=8080)

async def dm_on_github_async(userid,request):
    user = client.get_user(userid)
    if (user==None):
        abort(500, "User lookup failed");
    dm_channel = user.dm_channel
    if (dm_channel==None):
        dm_channel = await user.create_dm()
    if (dm_channel==None):
        abort(500, "Fail to create DM channel");
    # Handle the request
    event = request.get_header("X-GitHub-Event")
    await dm_channel.send("Got event "+str(event))
    #await dm_channel.send(str(request.body)) # Doesn't work well...
    return

@server.post("/dm_on_github/<userid:int>")
def dm_on_github(userid):
    request = bottle.request
    asyncio.run_coroutine_threadsafe(dm_on_github_async(userid,request.copy()),client.loop).result()


@client.event
async def on_ready():
    print('We have logged in as {0.user} '.format(client))
    # Wait for the old HTTP server
    if hasattr(client,"server_thread"):
        server.close()
        client.server_thread.join()
    client.server_thread = HTTPThread()
    client.server_thread.start()

#@client.event
#async def on_message(message):
#    if message.author == client.user:
#        return
#
#    if message.content.startswith('$hello'):
#        await message.channel.send('Hello!')

print("Starting Discord client")
client.run('super secret key')
print("Client terminated")
server.close()
print("Asked server to terminate")
server_thread.join()
print("Server thread successful join")