Starlette 异步 Web 框架¶

为什么要学 Starlette¶

Starlette 是一个轻量级的 ASGI 框架，是 FastAPI 的底层基础。它提供了路由、中间件、WebSocket、背景任务、静态文件等 Web 应用的核心功能，但不包含数据验证和 OpenAPI 文档生成。对于不需要 FastAPI 全部功能但想要高性能异步 Web 服务的场景（如微服务、WebSocket 服务、代理服务），Starlette 提供了最小化但完整的解决方案。

核心概念¶

概念	白话解释	用途
ASGI	异步服务器网关接口	Python 异步 Web 标准协议
Route	路由	URL 到处理函数的映射
Request/Response	请求/响应	HTTP 通信对象
Middleware	中间件	请求处理链
WebSocket	双向通信	实时数据交换
Background Task	后台任务	响应后异步执行的任务

安装配置¶

pip install starlette uvicorn
pip install httpx  # 测试客户端

快速上手¶

基本应用¶

from starlette.applications import Starlette
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles

async def homepage(request):
    return JSONResponse({"message": "Hello, Starlette!"})

async def user_detail(request):
    user_id = request.path_params["user_id"]
    return JSONResponse({"id": user_id, "name": "Alice"})

async def create_user(request):
    body = await request.json()
    return JSONResponse({"created": body}, status_code=201)

routes = [
    Route("/", homepage),
    Route("/users/{user_id:int}", user_detail),
    Route("/users", create_user, methods=["POST"]),
    Mount("/static", StaticFiles(directory="static"), name="static"),
]

app = Starlette(routes=routes)

uvicorn main:app --reload

中间件¶

from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
import time

class TimingMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            start = time.time()
            await self.app(scope, receive, send)
            print(f"Request took {time.time() - start:.3f}s")
        else:
            await self.app(scope, receive, send)

middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"]),
    Middleware(TimingMiddleware),
]

app = Starlette(routes=routes, middleware=middleware)

WebSocket¶

from starlette.routing import WebSocketRoute
from starlette.websockets import WebSocket

async def ws_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

routes = [
    WebSocketRoute("/ws", ws_endpoint),
]

进阶用法¶

后台任务¶

from starlette.background import BackgroundTask
from starlette.responses import JSONResponse

async def send_notification(email: str, message: str):
    # 耗时操作
    await asyncio.sleep(2)
    print(f"Notification sent to {email}")

async def create_order(request):
    body = await request.json()
    task = BackgroundTask(send_notification, body["email"], "订单已创建")
    return JSONResponse({"status": "created"}, background=task)

异常处理¶

from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse

async def http_exception_handler(request, exc):
    return JSONResponse(
        {"error": exc.detail},
        status_code=exc.status_code
    )

async def generic_exception_handler(request, exc):
    return JSONResponse(
        {"error": "Internal Server Error"},
        status_code=500
    )

app = Starlette(
    routes=routes,
    exception_handlers={
        HTTPException: http_exception_handler,
        Exception: generic_exception_handler,
    }
)

测试¶

from starlette.testclient import TestClient

def test_homepage():
    client = TestClient(app)
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, Starlette!"}

def test_create_user():
    client = TestClient(app)
    response = client.post("/users", json={"name": "Bob"})
    assert response.status_code == 201

生命周期事件¶

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    # 启动时
    print("Starting up...")
    db = await connect_to_db()
    app.state.db = db
    yield
    # 关闭时
    await db.close()
    print("Shutting down...")

app = Starlette(routes=routes, lifespan=lifespan)

常见问题¶

Q: Starlette 和 FastAPI 怎么选？¶

Starlette：更轻量、更底层、适合不需要自动验证和文档的场景
FastAPI：基于 Starlette + Pydantic，适合 REST API 开发

Q: 性能如何？¶

Starlette 是 Python 异步 Web 框架中最快的之一，FastAPI 的性能约等于 Starlette。

Q: 如何添加数据验证？¶

手动使用 Pydantic 或 marshmallow 验证。或者直接用 FastAPI。

参考资源¶

GitHub：https://github.com/encode/starlette
文档：https://www.starlette.io/
ASGI 规范：https://asgi.readthedocs.io/