你可以方便地使用 GraiaX FastAPI 配合 graia.amnesia.builtins.asgi.UvicornASGIService 轻松地在启动使用了 Graia Amnesia 的项目（如：Ariadne、Avilla）的同时启动一个 Uvicorn 服务器并在 Saya 模块中便捷地注册 FastAPI 的路由，且在 Launart 退出的时候自动关闭 Uvicorn。

安装

pdm add graiax-fastapi 或 poetry add graiax-fastapi。

我们强烈建议使用包管理器或虚拟环境

开始使用

以 Avilla 为例。

配合 Launart 使用

机器人入口文件

如果你有使用 Graia Saya 作为模块管理工具，那么你可以使用 FastAPIBehaviour 以在 Saya 模块中更方便地使用 FastAPI。

FastAPI 本身并 不自带 ASGI 服务器，因此你需要额外添加一个 UvicornASGIService。

import pkgutil

from avilla.console.protocol import ConsoleProtocol
from avilla.core import Avilla
from creart import create
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from graia.amnesia.builtins.asgi import UvicornASGIService
from graia.broadcast import Broadcast
from graia.saya import Saya
from launart import Launart

broadcast = create(Broadcast)
saya = create(Saya)
launart = create(Launart)
avilla = Avilla(broadcast=broadcast, launch_manager=launart, message_cache_size=0)
fastapi = FastAPI()

avilla.apply_protocols(ConsoleProtocol())
saya.install_behaviours(FastAPIBehaviour(fastapi))
fastapi.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
launart.add_component(FastAPIService(fastapi))
launart.add_component(UvicornASGIService("127.0.0.1", 9000, {"": fastapi}))  # type:ignore
# 上面这条命令会占据 Uvicorn 的所有入口点，详见下方的 Warning

with saya.module_context():
    for module in pkgutil.iter_modules(["modules"]):
        if module.name[0] in ("#", ".", "_"):
            continue
        saya.require(f"modules.{module.name}")

launart.launch_blocking()

Saya 模块中

from graia.saya import Channel
from pydantic import BaseModel

from graiax.fastapi import RouteSchema, route

channel = Channel.current()


class ResponseModel(BaseModel):
    code: int
    message: str


# 方式一：像 FastAPI 那样直接使用装饰器
@route.get("/", response_model=ResponseModel)
async def root():
    return {"code": 200, "message": "Hello World!"}


# 方式二：当你先需要同一个路径有多种请求方式时你可以这样做
@route.route(["GET"], "/xxxxx")
async def xxxxx():
    return "xxxx"


# 方式三：上面那种方式实际上也可以这么写
@channel.use(RouteSchema("/xxx", methods=["GET", "POST"]))
async def xxx():
    return "xxx"


# Websocket
from fastapi import WebSocket
from starlette.websockets import WebSocketDisconnect
from websockets.exceptions import ConnectionClosedOK


@route.ws("/ws")
# 等价于 @channel.use(WebsocketRouteSchema("/ws"))
async def ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        try:
            print(await websocket.receive_text())
        except (WebSocketDisconnect, ConnectionClosedOK, RuntimeError):
            break

其他方式

假如你不想在 Saya 模块中为 FastAPI 添加路由，那么你可以选择以下几种方式：

...
fastapi = FastAPI()
...
fastapi.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@fastapi.get("/main")
async def main():
    return "main"

...
launart.add_component(FastAPIService(fastapi))
launart.add_component(UvicornASGIService("127.0.0.1", 9000, {"": fastapi}))  # type:ignore
...

from fastapi.responses import PlainTextResponse
from avilla.standard.core.application.event import ApplicationReady
from graiax.fastapi.interface import FastAPIProvider

async def interface_test():
    return PlainTextResponse("I'm from interface!")


@listen(ApplicationReady)
async def function():
    launart = Launart.current()
    fastapi = launart.get_interface(FastAPIProvider)
    fastapi.add_api_route("/interface", fastapi.get("/interface")(interface_test))