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.
--reload:开发时自动重载(生产环境移除)访问:http://localhost:8000关键功能

(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.
适用场景构建高性能 REST API微服务后端实时应用(WebSocket)结合sqlalchemy进行 web 开发

阅读剩余
THE END