FastAPI学习之Starlette

FastAPI 与 Starlette 的关系

FastAPI 和 Starlette 是密切相关的 Python Web 框架，理解它们的关系有助于更好地选择和使用工具。

基本定义

• Starlette：
  • 轻量级ASGI（Asynchronous Server Gateway Interface）框架。
  • 提供基础的 HTTP 请求/响应处理、路由、中间件、WebSocket 支持等。
  • 核心代码约 3,000 行，无强制依赖。
• FastAPI：
  • 基于 Starlette 构建的高性能 Web 框架。
  • 在 Starlette 基础上添加了数据验证（通过 Pydantic）、依赖注入、自动 API 文档生成（Swagger/Redoc）等高级功能。

核心关系

• 继承关系：
  • FastAPI 直接继承 Starlette 的 Router 和 APIRoute，所有 Starlette 的功能在 FastAPI 中可用。
  • FastAPI 的 Request 和 Response 对象直接来自 Starlette。
• 扩展功能：

# FastAPI 扩展 Starlette 的示例
from fastapi import FastAPI
app = FastAPI() # 继承自 Starlette 的 Starlette 类

# 直接使用 Starlette 的中间件
from starlette.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware, allow_origins=["*"])

功能对比

• FastAPI = Starlette + Pydantic + 高级功能
• 若需要数据验证、依赖注入和自动化文档，优先选 FastAPI；若追求极简或深度定制，直接使用 Starlette。两者均可无缝协作。

性能对比

• 异步非阻塞：基于 async/await 语法，轻松处理 10K+ 并发连接
• 高效协议处理：内置 HTTP/1.1、HTTP/2、WebSocket 协议解析
• 底层性能一致：FastAPI 的性能几乎与 Starlette 相同，因为其核心依赖 Starlette 的异步处理。
• 测试数据（每秒请求数）：
  • Flask (WSGI): ~1,200 req/s
  • Starlette (ASGI): ~15,000 requests/s
  • FastAPI (基于 Starlette): ~14,900 requests/s（加入数据验证后仅有微小损耗）

使用场景

选择 Starlette 的情况

• 需要极简的 ASGI 框架。
• 开发高度定制化的中间件或协议（如自定义 WebSocket 逻辑）。
• 项目对依赖项大小敏感（Starlette 无强制依赖）。

选择 FastAPI 的情况

• 快速构建 RESTful API 或 Web 服务。
• 需要自动验证请求数据（如 JSON、表单）并生成文档。
• 复杂业务逻辑需要依赖注入解耦。

代码示例对比

Starlette 实现简单 API

from starlette.applications import Starlette
from starlette.responses import JSONResponse

app = Starlette()

@app.route("/")
async def homepage(request):
    return JSONResponse({"message": "Hello Starlette!"})

FastAPI 实现相同功能（带数据验证）

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Message(BaseModel):
    text: str

@app.post("/")
async def send_message(msg: Message):
    return {"response": f"Received: {msg.text}"}

如何混合使用

在 FastAPI 中可直接调用 Starlette 的功能：

from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()

# 直接使用 Starlette 的 WebSocket
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_text("Connected via Starlette!")

生态系统

• Starlette：
  • 生态较简单，适合作为底层工具。
  • 常用扩展：starlette-admin（后台管理）、starlette-exporter（Prometheus监控）。
• FastAPI：
  • 生态丰富，集成数据库（SQLAlchemy、TortoiseORM）、安全（OAuth2）、缓存等。
  • 热门插件：fastapi-users（用户管理）、fastapi-cache（缓存）。

Starlette深度解析

ASGI应用入口(app.py)

class Starlette:
    def __init__(self, routes=None, middleware=None, ...):
        self.router = Router(routes, ...)
        self.middleware_stack = self.build_middleware_stack(middleware)

    async def __call__(self, scope, receive, send):
        await self.middleware_stack(scope, receive, send)

    def build_middleware_stack(self, middleware):
        app = self.router
        for cls, options in reversed(middleware or []):
            app = cls(app=app, **options)
        return app

• 核心机制：通过中间件包装路由，形成处理链
• 执行流程：
  • 接收ASGI scope, receive, send
  • 中间件按添加顺序逆序包裹（类似洋葱模型）
  • 最终由Router 处理请求

路由系统(routing.py)

路由匹配逻辑

class Router:
    def __init__(self, routes):
        self.routes = []
        for route in routes:
            if isinstance(route, Route):
                self.add_route(route)
            elif isinstance(route, Mount):
                self.add_mount(route)

    async def handle(self, scope, receive, send):
        for route in self.routes:
            match, child_scope = route.matches(scope)
            if match == Match.FULL:
                scope.update(child_scope)
                await route.handle(scope, receive, send)
                return

• 匹配优先级：按添加顺序优先匹配
• 路径解析：使用parse 模块进行路径参数提取

路由类型

class Route:
    def __init__(self, path, endpoint, methods=None):
        self.path = path
        self.endpoint = endpoint #处理函数
        self.methods = methods

    async def handle(self, scope, receive, send):
        request = Request(scope, receive)
        response = await self.endpoint(request)
        await response(scope, receive, send)

请求/响应模型(requests.py / responses.py)

请求封装

class Request:
    def __init__(self, scope, receive):
        self.scope = scope
        self._receive = receive
        self._stream_consumed = False

    async def stream(self):
        while True:
            event = await self._receive()
            if event["type"] == "http.request":
                yield event.get("body", b"")
                if event.get("more_body", False) is False:
                    break

响应处理

class Response:
    def __init__(self, content, status_code=200, headers=None):
        self.body = self.render(content)
        self.status_code = status_code
        self.headers = Headers(headers or [])

    async def __call__(self, scope, receive, send):
        await send({
            "type": "http.response.start",
            "status": self.status_code,
            "headers": self.headers.raw,
        })
        await send({
            "type": "http.response.body",
            "body": self.body,
        })

中间件机制(middleware.py)

class Middleware:
    def __init__(self, app, **options):
        self.app = app
        self.options = options

    async def __call__(self, scope, receive, send):
        async def modified_send(message):
            #可修改响应消息
            await send(message)

        await self.app(scope, receive, modified_send)

典型中间件：

• CORSMiddleware：跨域处理
• GZipMiddleware：压缩响应
• HTTPSRedirectMiddleware：HTTPS重定向

WebSocket支持(websockets.py)

class WebSocket:
    async def accept(self, subprotocol=None):
        await self._send({"type": "websocket.accept", "subprotocol": subprotocol})

    async def receive(self):
        message = await self._receive()
        if message["type"] == "websocket.disconnect":
            raise WebSocketDisconnect(message["code"])
        return message

    async def send(self, data):
        if isinstance(data, str):
            await self._send({"type": "websocket.send", "text": data})
        else:
            await self._send({"type": "websocket.send", "bytes": data})

生命周期事件处理

class LifespanHandler:
    async def startup(self):
        for handler in self.on_startup:
            await handler()

    async def shutdown(self):
        for handler in self.on_shutdown:
            await handler()

    async def __call__(self, scope, receive, send):
        message = await receive()
        assert message["type"] == "lifespan.startup"
        await self.startup()
        await send({"type": "lifespan.startup.complete"})

        message = await receive()
        assert message["type"] == "lifespan.shutdown"
        await self.shutdown()
        await send({"type": "lifespan.shutdown.complete"})

性能优化设计

• 惰性加载：请求对象仅在需要时创建
• 零拷贝传输：大文件响应使用 FileResponse 的流式传输
• 高效头部处理：使用 MultiDict 数据结构存储头部
• 异步模板渲染：

class Jinja2Templates:
    async def TemplateResponse(self, name, context):
        template = self.get_template(name)
        content = template.render(context)
        return HTMLResponse(content)

核心设计模式

• 组合优于继承：通过中间件堆叠扩展功能
• 单一职责原则：每个类/函数专注单一功能
• 协议抽象：统一处理 HTTP/WebSocket 等协议
• 显式状态管理：严格区分客户端/服务器状态

通过这种模块化设计，Starlette 在保持代码精简（核心约 3,000 行）的同时，提供了高性能的 Web 开发基础。其设计哲学强调 “提供基础组件，而非完整解决方案”，这使得开发者可以灵活组合所需功能，而不被框架限制。

Starlette 扩展能力

生态系统

数据库集成

from databases import Database

database = Database("postgresql://user:pass@localhost/db")

@app.on_event("startup")
async def connect_db():
    await database.connect()

@app.on_event("shutdown")
async def disconnect_db():
    await database.disconnect()

模板引擎

from starlette.templating import Jinja2Templates

templates = Jinja2Templates(directory="templates")

async def user_profile(request):
    return templates.TemplateResponse(
        "profile.html", {"request": request, "user": "Alice"}
    )

后台任务

async def send_notification(email: str):
    # 模拟发送邮件
    print(f"Sending email to {email}")

async def create_user(request):
    background = request.state.background
    background.add_task(send_notification, "user@example.com")
    return JSONResponse({"status": "created"})

参考链接：

• Starlette
• GitHub – encode/starlette: The little ASGI framework that shines. 🌟
• The Core of FastAPI: A Deep Dive into Starlette 🌟🌟🌟 – DEV Community