asapi

asapi是FastAPI的一个轻量级且具有偏好的包装器。因为它是围绕FastAPI进行包装的，所以您可以将它整合到现有的项目中。

显式组合根

FastAPI使用Depends中的回调来执行其依赖项注入。这迫使您使用多层Depends来组合应用程序。这些Depends资源的创建通常分布在模块中，因此很难知道某物在哪里初始化。

FastAPI也没有应用程序级别的依赖项，因此您最终必须使用全局变量来在请求之间共享资源。

asapi通过具有显式组合根来解决此问题，您可以在一个地方定义所有依赖项。

端点然后使用Injected[DependencyType]来获取它们所需的依赖项。

示例

import anyio
from fastapi import FastAPI
from psycopg_pool import AsyncConnectionPool
from asapi import FromPath, Injected, serve, bind


app = FastAPI()


@app.get("/hello/{name}")
async def hello(
    name: FromPath[str],
    pool: Injected[AsyncConnectionPool],
) -> str:
    async with pool.connection() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT '¡Hola ' || %(name)s || '!'", {"name": name})
            res = await cur.fetchone()
            assert res is not None
            return res[0]

待办事项：将来，我希望提供围绕APIRouter和FastAPI的包装器，该包装器还强制您标记每个端点的每个参数为Injected、Query、Path、Body，这样就可以明确指出参数来自哪里，同时最小化样板代码。

在事件循环中运行

FastAPI建议使用Uvicorn来运行您的应用程序（注意：如果您使用Gunicorn，除非您正在部署到具有多个核心的“裸元”服务器，例如大型EC2实例，否则您可能不需要）。

但是，使用uvicorn app:app从命令行有几个问题

1. 它控制事件循环和启动权不在您的手中。您必须依赖于Uvicorn来配置事件循环、配置日志等。
2. 您必须使用ASGI生命周期来初始化资源，或者使用上面提到的全局技巧。
3. 您无法在事件循环中运行任何其他内容（例如后台工作进程）。

asapi通过提供一个serve函数来解决这一问题，您可以使用此函数在自己的事件循环中运行应用程序。

import anyio
from asapi import serve
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root() -> dict[str, str]:
    return {"message": "Hello World"}


async def main():
    await serve(app, 8000)

if __name__ == "__main__":
    anyio.run(main)

现在您完全控制事件循环，可以建立数据库连接、运行后台任务等。结合显式组合根，您可以在一个地方初始化所有资源。

import anyio
from fastapi import FastAPI
from psycopg_pool import AsyncConnectionPool
from asapi import FromPath, Injected, serve, bind


app = FastAPI()


@app.get("/hello/{name}")
async def hello(name: FromPath[str], pool: Injected[AsyncConnectionPool]) -> str:
    async with pool.connection() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT '¡Hola ' || %(name)s || '!'", {"name": name})
            res = await cur.fetchone()
            assert res is not None
            return res[0]


async def main() -> None:
    async with AsyncConnectionPool(
        "postgres://postgres:postgres@localhost:5432/postgres"
    ) as pool:
        bind(app, AsyncConnectionPool, pool)
        await serve(app, 8000)


if __name__ == "__main__":
    anyio.run(main)