目录:
Starlette 是 Python ASGI(异步服务器网关接口)框架,用于构建高性能异步 Web 应用,是 FastAPI 等流行框架的底层基础。
特点:
安装依赖:
$ pip install starlette -i https://pypi.tuna.tsinghua.edu.cn/simple
路由决定了 URL 与处理函数之间的映射关系。Starlette 通过 Route、Router、Mount 等类提供灵活的路由配置,支持路径参数和多种 HTTP 方法。
from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles
routes = [
Route('/', homepage),
Route('/users/{user_id:int}', user_detail),
Mount('/static', StaticFiles(directory='static')),
]
提供了友好的 Request 和 Response 对象抽象。Request 对象让开发者可以方便地访问方法、URL、头部、查询参数、路径参数、客户端地址、Cookies以及请求体(如表单、JSON、流)。响应则通过 JSONResponse、HTMLResponse、PlainTextResponse 等类简化。
from starlette.requests import Request
from starlette.responses import JSONResponse
async def read_user_agent(request: Request):
user_agent = request.headers.get('user-agent')
return JSONResponse({'user_agent': user_agent})
中间件是处理请求和响应的“拦截器”,如日志记录、CORS(跨域资源共享)、GZip压缩、身份验证等。Starlette 的中间件按顺序嵌套调用,形成“洋葱模型”。
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
app = Starlette(middleware=[Middleware(CORSMiddleware, allow_origins=['*'])])
原生支持 WebSocket 协议,便于构建实时双向通信应用,如聊天室、实时通知等。
from starlette.websockets import WebSocket
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
await websocket.send_text("Hello, WebSocket!")
data = await websocket.receive_text()
await websocket.close()
from starlette.applications import Starlette
from starlette.responses import JSONResponse, HTMLResponse
from starlette.routing import Route, Mount, WebSocketRoute
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from starlette.middleware.cors import CORSMiddleware
from starlette.websockets import WebSocket
import uvicorn
# 初始化模板引擎
templates = Jinja2Templates(directory="templates")
# 1. 定义HTTP请求处理函数
async def homepage(request):
"""首页,返回JSON响应"""
return JSONResponse({"message": "Hello from Starlette API!"})
async def html_page(request):
"""HTML页面,使用模板渲染"""
return templates.TemplateResponse("index.html", {"request": request, "title": "Starlette Demo"})
async def user_profile(request):
"""带路径参数的用户信息接口"""
user_id = request.path_params['user_id']
# 这里可以模拟异步数据库查询
return JSONResponse({'user_id': user_id, 'name': f'User{user_id}'})
# 2. 定义WebSocket处理函数
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Echo: {data}")
except Exception:
await websocket.close()
# 3. 定义路由
routes = [
Route("/", homepage),
Route("/welcome", html_page),
Route("/users/{user_id:int}", user_profile),
WebSocketRoute("/ws", websocket_endpoint),
Mount("/static", StaticFiles(directory="static"), name="static"),
]
# 4. 创建应用实例并添加中间件
app = Starlette(debug=True, routes=routes)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生产环境应指定具体域名
allow_methods=["*"],
allow_headers=["*"],
)
# 5. 运行应用(使用Uvicorn ASGI服务器)
if __name__ == "__main__":
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)
↶ 返回首页 ↶