当前位置：首页 > news >正文

fastapi

news 2025/9/18 11:04:57

FastAPI 框架快速学习

概述

FastAPI 是一个用于构建 API 的现代、高性能 Python Web 框架，基于 Starlette（Web 服务器）和 Pydantic（数据验证）。

核心特性：

极高性能：得益于 Starlette 和 Pydantic，性能与 NodeJS 和 Go 相当
高效开发：自动交互式文档、类型提示支持，提升开发速度 200-300%
更少错误：类型系统减少约 40% 的人为错误
生产就绪：自动生成 OpenAPI 文档，支持认证、中间件等

一、快速开始

安装与基础使用

pip install fastapi uvicorn

from fastapi import FastAPIapp = FastAPI()@app.get("/")
async def root():return {"message": "Hello World"}# 直接运行
if __name__ == "__main__":import uvicornuvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

运行后访问：

http://127.0.0.1:8000 - API 端点
http://127.0.0.1:8000/docs - 自动交互式文档

二、路径操作

支持多种 HTTP 方法

from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id: int):return {"item_id": item_id}@app.post("/items/")
async def create_item(item: dict):return {"item": item}@app.put("/items/{item_id}")
async def update_item(item_id: int, item: dict):return {"item_id": item_id, "updated_item": item}@app.delete("/items/{item_id}")
async def delete_item(item_id: int):return {"message": f"Item {item_id} deleted"}

路由分组与组织

from fastapi import APIRouteruser_router = APIRouter(prefix="/users", tags=["Users"])@user_router.get("/")
async def get_users():return [{"id": 1, "name": "John"}]@user_router.get("/{user_id}")
async def get_user(user_id: int):return {"id": user_id, "name": "John"}# 主应用中注册路由
from fastapi import FastAPI
app = FastAPI()
app.include_router(user_router)

三、请求处理

路径参数与查询参数

from typing import Optional@app.get("/users/{user_id}")
async def read_user(user_id: int,  # 路径参数skip: int = 0,  # 查询参数，有默认值limit: Optional[int] = 10,  # 可选查询参数search: Optional[str] = None  # 可选搜索参数
):return {"user_id": user_id, "skip": skip, "limit": limit, "search": search}

请求体与 Pydantic 模型

from pydantic import BaseModel, EmailStr
from typing import List, Optionalclass Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattags: List[str] = []class User(BaseModel):email: EmailStrpassword: str@app.post("/items/")
async def create_item(item: Item):return item@app.post("/users/")
async def create_user(user: User):# 密码应该哈希处理，这里仅为示例return {"email": user.email, "message": "User created"}

表单数据处理

from fastapi import Form@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):return {"username": username}

文件上传

from fastapi import UploadFile, File
from typing import List@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):return {"filename": file.filename}@app.post("/upload-multiple/")
async def upload_multiple_files(files: List[UploadFile] = File(...)):return {"filenames": [file.filename for file in files]}

四、响应处理

响应模型与过滤

class UserOut(BaseModel):id: intemail: EmailStrfull_name: Optional[str] = Noneclass UserIn(BaseModel):email: EmailStrpassword: strfull_name: Optional[str] = None@app.post("/users/", response_model=UserOut)
async def create_user(user: UserIn):# 在实际应用中，这里会创建用户并返回用户数据return user  # 自动过滤掉密码字段# 排除未设置的字段
@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):return items[item_id]  # 只返回实际设置的字段

五、数据库集成

Tortoise ORM 配置

from tortoise.contrib.fastapi import register_tortoiseTORTOISE_ORM = {"connections": {"default": "sqlite://db.sqlite3"},"apps": {"models": {"models": ["models", "aerich.models"],"default_connection": "default",}},
}register_tortoise(app,config=TORTOISE_ORM,generate_schemas=True,  # 自动创建表add_exception_handlers=True,
)

模型定义与 CRUD 操作

from tortoise.models import Model
from tortoise import fieldsclass User(Model):id = fields.IntField(pk=True)email = fields.CharField(max_length=255, unique=True)hashed_password = fields.CharField(max_length=255)created_at = fields.DatetimeField(auto_now_add=True)# 创建用户
user = await User.create(email="test@example.com", hashed_password="hashed_pw")# 查询用户
user = await User.get(email="test@example.com")
users = await User.all().values("id", "email")# 更新用户
await User.filter(id=1).update(email="new@example.com")# 删除用户
await User.filter(id=1).delete()

六、中间件

自定义中间件

from fastapi import Request
import time@app.middleware("http")
async def add_process_time_header(request: Request, call_next):start_time = time.time()response = await call_next(request)process_time = time.time() - start_timeresponse.headers["X-Process-Time"] = str(process_time)return response

关于 @app.middleware("http")：

"http" 表示这是一个 HTTP 中间件，处理 HTTP 请求/响应周期
其他可选类型包括 "websocket"（处理 WebSocket 连接）
中间件按照添加顺序执行，响应时按相反顺序处理

CORS 中间件

from fastapi.middleware.cors import CORSMiddlewareapp.add_middleware(CORSMiddleware,allow_origins=["http://localhost:3000"],  # 前端地址allow_credentials=True,allow_methods=["*"],  # 允许所有方法allow_headers=["*"],  # 允许所有头
)

七、高级特性

依赖注入

from fastapi import Depends, Header, HTTPExceptionasync def get_token_header(x_token: str = Header(...)):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")return x_token@app.get("/items/", dependencies=[Depends(get_token_header)])
async def read_items():return [{"item": "Foo"}, {"item": "Bar"}]

后台任务

from fastapi import BackgroundTasksdef write_notification(email: str, message=""):# 这里可以发送邮件或进行其他异步操作with open("log.txt", mode="w") as email_file:content = f"notification for {email}: {message}"email_file.write(content)@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):background_tasks.add_task(write_notification, email, message="some notification")return {"message": "Notification sent in the background"}

静态文件服务

from fastapi.staticfiles import StaticFilesapp.mount("/static", StaticFiles(directory="static"), name="static")

八、错误处理

自定义异常处理器

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponseclass CustomException(Exception):def __init__(self, name: str):self.name = name@app.exception_handler(CustomException)
async def custom_exception_handler(request, exc):return JSONResponse(status_code=418,content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},)@app.get("/custom-exception/")
async def raise_custom_exception():raise CustomException("Custom error")

最佳实践

项目结构：

my_project/
├── main.py
├── routers/
│   ├── __init__.py
│   ├── items.py
│   └── users.py
├── models/
│   ├── __init__.py
│   └── models.py
└── schemas/├── __init__.py└── schemas.py