使用FastAPI构建高性能RESTful API的完整指南_开发_开发者

引言

FastAPI是一个现代、高性能的python Web框架,专为构建API而设计。它基于标准的Python类型提示,提供了自动的API文档生成、数据验证和序列化等强大功能。本指南将带你从基础到高级,全面掌握FastAPI的使用。

为什么选择FastAPI

FastAPI相比其他框架有以下显著优势:

高性能: 性能可与Nodejs和Go媲美,是最快的Python框架之一
开发效率高: 比传统框架减少约40%的开发时间
自动文档: 自动生成交互式API文档(Swagger UI和ReDoc)
类型安全: 基于Python类型提示,提供编辑器自动补全和类型检查
标准化: 基于OpenAPI和JSON Schema标准
异步支持: 原生支持async/await语法

环境搭建

安装依赖

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装FastAPI和ASGI服务器
pip install fastapi
pip编程客栈 install "uvicorn[standard]"

第一个API应用

创建main.py文件:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

运行应用:

uvicorn main:app --reload

访问http://localhost:8000即可看到返回结果,访问http://localhost:8000/docs可以看到自动生成的交互式文档。

核心概念

路径参数

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

FastAPI会自动进行类型转换和验证,如果item_id不是整数,会返回清晰的错误信息。

查询参数

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

查询参数在URL中以?skip=0&limit=10的形式传递。

请求体

使用Pydantic模型定义请求体:

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/")
async def create_item(item: Item):
    return item

Pydantic会自动验证请求数据,并提供清晰的错误提示。

数据验证

使用Field进行高级验证

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0, description="价格必须大于0")
    quantity: int = Field(default=1, ge=1, le=1000)
    tags: list[str] = Field(default_factory=list)

使用Query、Path和Body

from fastapi import Query, Path, Body

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., title="商品ID", ge=1),
    q: str | None = Query(None, min_length=3, max_length=50),
    size: int = Query(default=10, le=100)
):
    return {"item_id": item_id, "q": q, "size": size}

响应模型

定义响应结构

class ItemResponse(BaseModel):
    id: int
    name: str
    price: float
    
    class Config:
        from_attributes = True

@app.post("/items/", response_model=ItemResponse)
async def create_item(item: Item):
    # 假设保存到数据库后返回
    return {"id": 1, **item.dict()}

多种响应状态码

from fastapi import status

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    return None

错误处理

抛出HTTP异常

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(
            status_code=404,
            detail="商品未找到",
            headers={"X-Error": "Item not found"}
        )
    return items_db[item_id]

自定义异常处理器

from fastapi.responses import JSONResponse
from fastapi import Request

class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"处理 {exc.name} 时出错"}
    )

依赖注入

基础依赖

from fastapi import Depends

async def common_parameters(q: str | None = 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_parameters)):
    return commons

类作为依赖

class Pagination:
    def __init__(self, skip: int = 0, limit: int = 100):
        self.skip = skip
        self.limit = limit

@app.get("/users/")
async def read_users(pagination: Pagination = Depends()):
    return {"skip": pagination.skip, "limit": pagination.limit}

数据库会话依赖

from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/")
async def read_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users

认证和授权

JWT认证示例

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Security
import jwt

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="无效的token")

@app.get("/protected/")
async def protected_route(payload: dict = Depends(verify_token)):
    return {"user": payload["sub"]}

OAuth2密码流

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # 验证用户名和密码
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="用户名或密码错误")
    
    Access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

数据库集成

SQLAlchemy集成

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "SQLite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    name = Column(String)

Base.metadata.create_all(bind=engine)

@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

中间件

添加CORS中间件

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

自定义中间件

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_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

后台任务

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message + "\n")

@app.post("/send-notification/")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(write_log, f"通知发送到: {email}")
    return {"message": "通知将在后台发送"}

文件上传和下载

文件上传

from fastapi import File, UploadFile

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents)
    }

多文件上传

@app.poKnAIfvst("/uploadfiles/")
async def create_upload_files(files: list[UploadFile] = File(...)):
    return [{"filename": file.filename} for file in files]

测试

使用TestClient

from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "测试商品", "price": 99.99}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "测试商品"

性能优化

使用异步操作

import httpx

@app.get("/external-api/")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

数据库查询优化

from sqlalchemy.orm import joinedload

@app.get("/users/{user_id}/posts/")
def get_user_posts(user_id: int, db: Session = Depends(get_db)):
    # 使用eager loading避免N+1查询问题
    user = db.query(User).options(
        joinedload(User.posts)
    ).filter(User.id == user_id).first()
    return user.posts

响应缓存

from functools import lru_cache

@lru_cache(maxsize=128)
def get_settings():
    return Settings()

@app.get("/settings/")
def read_settings(settings: Settings = Depends(get_settings)):
    return settings

部署

使用Docker

创建Dockerfile:

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

生产环境配置

# 使用Gunicorn + Uvicorn workers
# 命令行运行:
# gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

环境变量配置

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI应用"
    database_url: str
    secret_key: str
    
    class Config:
        env_file = ".env"

settings = Settings()

最佳实践

项目结构

project/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── models/
│ │ ├── __init__.py
│ │ └── user.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ └── user.py
│ ├── routers/
│ │ ├── __init__.py
│ │ └── users.py
│ ├── dependencies.py
│ └── database.py
├── tests/
├── requirements.txt
└── .env

使用APIRouter组织路由

from fastapi import APIRouter

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"descripjstion": "Not found"}}
)

@router.get("/")
async def read_users():
    return [{"username": "user1"}]

@router.get("/{user_id}")
async def read_user(user_id: int):
    return android{"user_id": user_id}

# 在main.py中
app.include_router(router)

版本控制

from fastapi import APIRouter

api_v1 = APIRouter(prefix="/api/v1")
api_v2 = APIRouter(prefix="/api/v2")

@api_v1.get("/items/")
async def get_items_v1():
    return {"version": "1.0"}

@api_v2.get("/items/")
async def get_items_v2():
    return {"version": "2.0", "data": []}

app.include_router(api_v1)
app.include_router(api_v2)

总结

FastAPI是一个强大而现代的Python Web框架,它结合了高性能、开发效率和优秀的开发者体验。通过本指南,你已经学习了:

FastAPI的核心概念和基础用法
数据验证和序列化
依赖注入系统
认证和授权
数据库集成
性能优化技巧
生产环境部署

继续深入学习FastAPI的官方文档,并在实际项目中应用这些知识,你将能够构建出高性能、可维护的RESTful API应用。

以上就是使用FastAPI构建高性能RESTful API的完整指南的详细内容，更多关于FastAPI构建高性能API的资料请关注编程客栈(www.devze.com)其它相关文章！

使用FastAPI构建高性能RESTful API的完整指南

目录

引言