FastAPI, 一个神奇的 Python 库
FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,使用 Python 并基于标准的 Python类型提示。
安装 FastAPI 很简单,这里我们使用 pip 命令来安装。
复制
pip install fastapi1.
另外我们还需要一个 ASGI 服务器,生产环境可以使用 Uvicorn 或者 Hypercorn
复制
pip install uvicorn[standard]1.
资料:
FastAPI 文档:https://fastapi.tiangolo.com/zh/FastAPI 源码:https://github.com/tiangolo/fastapi简单示例复制
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}1.2.3.4.5.6.7.8.9.
启动服务:
复制
uvicorn main:app --reload1.
(1) 路径参数和查询参数
{user_id}:路径参数(必须)limit 和 skip:查询参数(可选,带默认值)复制
@app.get("/users/{user_id}")
async def read_user(user_id: int,
limit: int = 10,
skip: int = 0):
return {"user_id": user_id, "limit": limit, "skip": skip}1.2.3.4.5.
(2) 请求体(POST/PUT)
使用 Pydantic 模型验证数据:
复制
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.post("/items/")
async def create_item(item: Item):
return {"item": item.dict()}1.2.3.4.5.6.7.8.9.10.11.
(3) 自动文档
Swagger UI:http://localhost:8000/docsReDoc:http://localhost:8000/redoc(4) 异步支持
复制
@app.get("/slow-endpoint")
async def slow_operation():
# 如数据库查询
await some_async_task()
return {"status": "done"}1.2.3.4.5.
(1) 依赖注入
复用代码逻辑(如认证、数据库连接)。
复制
from fastapi import Depends
def common_params(q: str = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_params)):
return commons1.2.3.4.5.6.7.8.
(2) 中间件
处理请求前/后的逻辑(如 CORS、日志)。
复制
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
)1.2.3.4.5.6.7.
(3) WebSocket 支持
复制
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message: {data}")1.2.3.4.5.6.
(4) 后台任务
执行无需即时响应的操作(如发送邮件、短信)
复制
from fastapi import BackgroundTasks
def log_task(message: str):
with open("log.txt", "a") as f:
f.write(message)
@app.post("/send-email")
async def send_email(background_tasks: BackgroundTasks):
background_tasks.add_task(log_task, "Email sent")
return {"status": "ok"}1.2.3.4.5.6.7.8.9.10.
阅读剩余
THE END