Как запустить uvicorn не блокируя основной поток?

Question

[Tiubert]

Делаю что-то на подобии dashboard для бота который запускается вместе с ботом в одной консоли. Логика дашбоарда вынесена в отдельный файл dashboard.py, бот запускается в main.py и импортирует класс Dashboard где лежит метод run_host - он принимает пару аргументов для примера и запускает хост сайта на локали с помощью uvicorn.run. Запускается при чем этот метод(синхронный) в отдельном потоке чтобы не блокировать основной т.к. там работает бот.

Ну и в общем оно работает, да, но чтобы вызывать асинхронные методы и функции при переходе на сайт(это к примеру функция read_root) вместо обычного await somefunc() нужно писать asyncio.run_coroutine_threadsafe(somefunc(), self.loop).result() потому что при обычном вызове мы получаем RuntimeError: Timeout context manager should be used inside a task

Как запустить uvicorn чтобы не пришлось использовать метод asyncio.run_coroutine_threadsafe?

main.py

import asyncio
from threading import Thread

import discord
from discord.ext import commands

from dashboard import Dashboard

bot = commands.Bot(command_prefix="!", intents=discord.Intents.all())

@bot.event
async def on_ready():
    dashboard = Dashboard()
    Thread(target=dashboard.run_host, args=(bot, asyncio.get_event_loop())).start()

@bot.event
async def on_message(message):
    if message.content == "!hello":
        await message.channel.send("Hello")

if __name__ == "__main__":
    bot.run("token")

dashboard.py

import asyncio

import uvicorn
from fastapi import FastAPI, Request
from starlette.templating import Jinja2Templates


class Dashboard:
    def __init__(self):
        self.client = None
        self.app = FastAPI()
        self.templates = Jinja2Templates(directory="templates")

        @self.app.get("/")
        async def read_root(request: Request):
            info = asyncio.run_coroutine_threadsafe(self.client.application_info(), self.loop).result()

            return self.templates.TemplateResponse("index.html", {
                "request": request,
                "name": info.name,
            })

    def run_host(self, client, loop):
        self.client = client
        self.loop = loop
        uvicorn.run(self.app, host="localhost", port=8000, log_level="error")

Что пытался сделать

Кроме потоков пытался запустить uvicorn с помощью мультипроцессинга, но FastAPI выдает ошибку:
AttributeError: Can't pickle local object 'FastAPI.setup..openapi'
По сути в коде он ничем не отличался от потокового выполнения:
mp.Process(dashboard.run_host, args=[bot, asyncio.get_event_loop()])
Можно просто строчку заменить.

Нашел такой вопрос (Как запустить несколько ботов в докере?) где автор поделился способом asyncio.get_event_loop().create_task, но тоже не работает - RuntimeError: asyncio.run() cannot be called from a running event loop. Эта ошибка к тому же относится к причине почему я добавил библиотеку discord.py к вопросу, выполнение кода происходит внутри asyncio.run и из него вызывается тот самый uvicorn.run

Я пробовал все виды loop параметра у uvicorn, кроме uvloop т.к. не установлен модуль, при использовании мультипроцессинга и многопоточности.

Не обязательно запускать dashboard в той же консоли что и основной поток, нужна только передача некоторых аргументов в том числе python объектов при запуске (в коде наглядный пример запуска).

Answer 1

[Vindicar]

Возможно, придётся заглянуть с другой стороны. uvicorn вроде позволяет запускать корутины при старте сервера? Оформи бота как корутину и используй этот механизм, чтобы uvicorn запускал бота, а не наоборот.

Comments

[Tiubert]

Тяжело будет такое реализовать когда сервер является модулем для бота, ну посмотрим что выйдет.

[Tiubert]

Я добился того же результата. По прежнему ошибка RuntimeError: Timeout context manager should be used inside a task одна использую startup ивент от FastAPI.

В main.py убрал запуск бота и добавил это

if __name__ == "__main__":
    dashboard = Dashboard()
    dashboard.run_host(bot, asyncio.get_event_loop())

on_ready естественно почистил

Это рабочий вариант

class Dashboard:
    def __init__(self):
        self.client = None
        self.app = FastAPI()
        self.templates = Jinja2Templates(directory="templates")

        @self.app.get("/")
        async def read_root(request: Request):
            info = asyncio.run_coroutine_threadsafe(self.client.application_info(), self.loop).result()

            return self.templates.TemplateResponse("index.html", {
                "request": request,
                "name": info.name,
            })

        @self.app.on_event("startup")
        async def on_startup():
            Thread(target=self.client.run, args=("token",)).start()

Ну а так уже не будет работать и будет ошибка

- info = asyncio.run_coroutine_threadsafe(self.client.application_info(), self.loop).result()
+ info = self.client.application_info() # если заменить одну строчку

В общем мы ничего не добились и ничего от этого не поменялось, хотя может быть я не туда копаю.