wehub-skill-sync
126 lines
4.0 KiB
Markdown
126 lines
4.0 KiB
Markdown
# Runnable Example: FastAPI / Web Server
|
||
|
||
**当应用是 Web 服务器时**（FastAPI、Flask、Starlette），需要测试完整的 HTTP 请求管道。
|
||
|
||
**方法**：使用 `httpx.AsyncClient` 配合 `ASGITransport`，在进程内运行 ASGI 应用。这是最快且最可靠的方式——无需子进程，无需管理端口。
|
||
|
||
```python
|
||
# 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()` 中手动复制该初始化逻辑：
|
||
|
||
```python
|
||
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、基于文件的数据库、全局缓存），请添加信号量来串行化访问：
|
||
|
||
```python
|
||
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 访问：
|
||
|
||
```python
|
||
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
|
||
bash resources/run-with-timeout.sh 120 uv run python -m myapp.server
|
||
sleep 3 # 等待就绪
|