skillhub-048-eval-driven-dev/references/runnable-examples/fastapi-web-server.md

Runnable Example: FastAPI / Web Server

当应用是 Web 服务器时（FastAPI、Flask、Starlette），需要测试完整的 HTTP 请求管道。

方法：使用 httpx.AsyncClient 配合 ASGITransport，在进程内运行 ASGI 应用。这是最快且最可靠的方式——无需子进程，无需管理端口。

# pixie_qa/run_app.py
import httpx
from pydantic import BaseModel
import pixie


class AppArgs(BaseModel):
    user_message: str


class AppRunnable(pixie.Runnable[AppArgs]):
    """通过进程内 ASGI 传输驱动 FastAPI 应用。"""

    _client: httpx.AsyncClient

    @classmethod
    def create(cls) -> "AppRunnable":
        return cls()

    async def setup(self) -> None:
        from myapp.main import app  # 你的 FastAPI/Starlette 应用实例

        transport = httpx.ASGITransport(app=app)
        self._client = httpx.AsyncClient(transport=transport, base_url="http://test")

    async def run(self, args: AppArgs) -> None:
        await self._client.post("/chat", json={"message": args.user_message})

    async def teardown(self) -> None:
        await self._client.aclose()

ASGITransport 会跳过生命周期事件

httpx.ASGITransport 不会触发 ASGI 生命周期事件（startup / shutdown）。如果应用在其生命周期中初始化了资源（数据库连接、缓存、服务客户端），则必须在 setup() 中手动复制该初始化逻辑：

async def setup(self) -> None:
    # 手动复制应用生命周期中的操作
    from myapp.db import get_connection, init_db, seed_data
    import myapp.main as app_module

    conn = get_connection()
    init_db(conn)
    seed_data(conn)
    app_module.db_conn = conn  # 设置应用期望的模块级全局变量

    transport = httpx.ASGITransport(app=app_module.app)
    self._client = httpx.AsyncClient(transport=transport, base_url="http://test")

async def teardown(self) -> None:
    await self._client.aclose()
    # 清理手动初始化的资源
    import myapp.main as app_module
    if hasattr(app_module, "db_conn") and app_module.db_conn:
        app_module.db_conn.close()

共享可变状态下的并发

如果应用使用了共享可变状态（内存 SQLite、基于文件的数据库、全局缓存），请添加信号量来串行化访问：

import asyncio

class AppRunnable(pixie.Runnable[AppArgs]):
    _client: httpx.AsyncClient
    _sem: asyncio.Semaphore

    @classmethod
    def create(cls) -> "AppRunnable":
        inst = cls()
        inst._sem = asyncio.Semaphore(1)
        return inst

    async def setup(self) -> None:
        from myapp.main import app
        transport = httpx.ASGITransport(app=app)
        self._client = httpx.AsyncClient(transport=transport, base_url="http://test")

    async def run(self, args: AppArgs) -> None:
        async with self._sem:
            await self._client.post("/chat", json={"message": args.user_message})

    async def teardown(self) -> None:
        await self._client.aclose()

仅在需要时才使用信号量——如果应用使用以唯一 ID（call_sid、session_id）为键的按会话状态，则并发调用天然隔离，无需加锁。

备选方案：外部服务器配合 httpx

当应用无法直接导入时（复杂的启动流程、__main__ 中的 uvicorn.run()），以子进程方式启动并通过 HTTP 访问：

class AppRunnable(pixie.Runnable[AppArgs]):
    _client: httpx.AsyncClient

    @classmethod
    def create(cls) -> "AppRunnable":
        return cls()

    async def setup(self) -> None:
        # 假定服务器已在运行（通过 run-with-timeout.sh 启动）
        self._client = httpx.AsyncClient(base_url="http://localhost:8000")

    async def run(self, args: AppArgs) -> None:
        await self._client.post("/chat", json={"message": args.user_message})

    async def teardown(self) -> None:
        await self._client.aclose()

在运行 pixie trace 或 pixie test 之前启动服务器：

bash resources/run-with-timeout.sh 120 uv run python -m myapp.server
sleep 3  # 等待就绪