【Python教程】FastAPI的高级特性与最佳实践

FastAPI 是一个现代化的、快速（高性能）的 Web 框架，基于 Python 3.6+，使用标准的 Python 类型提示。它不仅简单易用，还提供了许多高级特性，使得构建高性能的 API 变得更加容易。本篇文章将详细介绍 FastAPI 的一些高级特性，并分享一些最佳实践。文章源自灵鲨社区-https://www.0s52.com/bcjc/pythonjc/15937.html

1. 依赖注入

依赖注入是一种设计模式，可以使代码更加模块化和可测试。FastAPI 通过 Depends 实现依赖注入。文章源自灵鲨社区-https://www.0s52.com/bcjc/pythonjc/15937.html

1.1 基本用法

from fastapi import Depends, FastAPI

app = FastAPI()

def common_parameters(q: str = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

• 定义了一个函数 common_parameters，它接受三个参数，并返回一个字典。
• 使用 Depends 将 common_parameters 函数注入到 read_items 路由中。

1.2 深入理解 Depends

Depends 可以接受任何可调用对象，如函数或类，并自动解决其依赖关系。这使得代码更具模块化和可复用性。文章源自灵鲨社区-https://www.0s52.com/bcjc/pythonjc/15937.html

from fastapi import Depends, HTTPException, status

def verify_token(token: str):
    if token != "supersecrettoken":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
        )
    return token

@app.get("/protected/")
async def read_protected(token: str = Depends(verify_token)):
    return {"message": "This is a protected endpoint"}

• 定义了一个 verify_token 函数，用于验证传入的 token。
• 在路由中使用 Depends 将 verify_token 注入，确保请求时提供有效的 token。

2. 中间件

中间件是一些在请求处理过程开始和结束时执行的函数。FastAPI 支持基于 ASGI 的中间件。文章源自灵鲨社区-https://www.0s52.com/bcjc/pythonjc/15937.html

2.1 基本中间件

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class SimpleMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers['X-Custom-Header'] = 'Value'
        return response

app.add_middleware(SimpleMiddleware)

• 定义了一个 SimpleMiddleware 类，继承自 BaseHTTPMiddleware。
• 实现 dispatch 方法，在请求处理前后执行代码。
• 使用 app.add_middleware 添加中间件。

文章源自灵鲨社区-https://www.0s52.com/bcjc/pythonjc/15937.html

2.2 内置中间件

FastAPI 提供了一些内置中间件，例如 CORSMiddleware 用于处理跨域请求。文章源自灵鲨社区-https://www.0s52.com/bcjc/pythonjc/15937.html

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

• 使用 add_middleware 添加内置的 CORSMiddleware。
• 配置允许的来源、方法和头部，允许所有跨域请求。

3. 事件处理

FastAPI 支持在应用启动和关闭时执行一些特定的事件处理函数。

3.1 启动和关闭事件

@app.on_event("startup")
async def startup_event():
    print("Application startup")

@app.on_event("shutdown")
async def shutdown_event():
    print("Application shutdown")

• 使用 @app.on_event 装饰器注册启动和关闭事件。
• 定义在应用启动时执行的 startup_event 函数，以及在应用关闭时执行的 shutdown_event 函数。

4. 依赖注入系统

FastAPI 的依赖注入系统不仅可以简化代码，还可以提高测试和复用性。

4.1 类作为依赖项

class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 10):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    return commons

• 定义一个 CommonQueryParams 类，包含一些查询参数。
• 在路由中使用 Depends 注入类的实例。

4.2 依赖项的生命周期

可以在依赖项中执行一些资源初始化和清理操作。

DATABASE_URL = "mysql+aiomysql://root:123456@localhost/test_db"


async def get_db():
    db = Database(DATABASE_URL)
    print("1. 连接数据库")
    await db.connect()
    try:
        yield db
    finally:
        print("3. 断开数据库连接")
        await db.disconnect()


@app.get("/items/")
async def read_items(db: Database = Depends(get_db)):
    query = "SELECT * FROM users"
    print("2. 执行查询")
    items = await db.fetch_all(query)
    return items

• 定义了一个 get_db 函数，使用 yield 提供数据库连接，并在请求完成后关闭连接。
• 在路由中使用 Depends 注入数据库连接。

5. 认证与授权

FastAPI 支持多种认证和授权方式，以下是使用 OAuth2 的示例。

5.1 OAuth2 密码模式

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    if token != "fake-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"username": "fake-user"}

@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

• 使用 OAuth2PasswordBearer 定义一个依赖项 oauth2_scheme，指定获取 token 的 URL。
• 定义 get_current_user 函数，验证 token 并返回当前用户信息。
• 在路由中使用 Depends 注入 get_current_user，确保请求时提供有效的 token。

6. 异步编程与性能优化

FastAPI 是为高性能设计的，充分利用了 Python 的异步功能。

6.1 异步请求处理

import asyncio
from fastapi import FastAPI

app = FastAPI()

async def task1():
    await asyncio.sleep(1)
    return "Task 1 completed"

async def task2():
    await asyncio.sleep(2)
    return "Task 2 completed"

@app.get("/tasks/")
async def run_tasks():
    results = await asyncio.gather(task1(), task2())
    return results

• 定义了两个异步任务 task1 和 task2。
• 使用 asyncio.gather 并行运行这两个任务，并返回结果。

6.2 利用缓存

当使用 FastAPI 开发应用程序时，为了提高性能和降低响应时间，可以使用缓存技术。主要的缓存方案之一是使用 Redis，一个高效的内存键值存储数据库。

from fastapi import FastAPI
import json
import redis

app = FastAPI()
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

@app.get("/data")
def read_data():
    cached_data = redis_client.get("cached_data")
    if cached_data:
        return json.loads(cached_data)

    # 模拟生成数据的耗时操作
    data = {"message": "Hello, World!"}
    redis_client.setex("cached_data", 3600, json.dumps(data))  # 缓存时间为1小时

    return data

实现缓存的好处

• 降低计算成本：避免重复执行耗时的计算或数据库查询。
• 提高响应速度：从内存中读取数据比从数据库或计算中获取要快得多。
• 缓解数据库负载：减少对数据库的频繁访问，提高系统的整体性能。

7. 数据库集成与操作

FastAPI 可以轻松地与数据库集成，以下是一个与MySQL集成的示例。

7.1 数据库连接

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship

# 定义数据库连接 URL
DATABASE_URL = "mysql+pymysql://root:123456@localhost/test_db"

# 创建数据库引擎
engine = create_engine(DATABASE_URL)

# 声明一个基类，用于定义 ORM 映射
Base = declarative_base()


# 定义 Class 类，映射到数据库中的 classes 表
class Class(Base):
    __tablename__ = "classes"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    description = Column(String(255))
    teacher_id = Column(Integer, ForeignKey('teachers.id'))

    # 定义关系，反向映射到 Teacher 类的实例
    teacher = relationship("Teacher", back_populates="classes")


class Teacher(Base):
    __tablename__ = "teachers"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), unique=True, index=True)

    # 定义关系，反向映射到 Class 类的实例
    classes = relationship("Class", back_populates="teacher")


# 创建数据库表结构
Base.metadata.create_all(bind=engine)

解释：

• 使用 create_engine 创建与 MySQL 数据库的连接。
• 定义 Class 和 Teacher 模型类，映射到数据库中的 classes 和 teachers 表。
• 调用 Base.metadata.create_all 创建表。

7.2 数据库操作

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

DATABASE_URL = "mysql+pymysql://root:123456@localhost/test_db"

# 创建数据库引擎
engine = create_engine(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)
    name = Column(String(50), index=True)
    email = Column(String(100), unique=True, index=True)


# 创建数据库表
Base.metadata.create_all(bind=engine)


# 定义Pydantic模型
class UserCreate(BaseModel):
    name: str
    email: str


class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        from_attributes = True


app = FastAPI()


# 数据库依赖项
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


@app.get("/users/{user_id}", response_model=UserResponse)
async def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

解释：

• 定义 get_db 依赖项，用于获取数据库会话。
• 创建用户的路由，接受用户数据，插入数据库并返回。
• 查询用户的路由，根据用户 ID 查询并返回用户信息。

8. 测试与调试

测试和调试是开发过程中的重要环节，确保代码的正确性和稳定性。

8.1 使用 pytest 进行测试

from fastapi.testclient import TestClient
from .main import app

client = TestClient(app)

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

解释：

• 使用 TestClient 创建测试客户端。
• 定义测试函数 test_read_main，发送GET请求并断言响应状态码和内容。

8.2 使用 pdb 调试

在代码中插入断点，可以使用 pdb 进行调试。

import pdb; pdb.set_trace()

# 代码

解释：

• 在需要调试的地方插入 pdb.set_trace()。
• 运行代码后程序会在断点处暂停，进入调试模式，可以逐行执行代码并检查变量值。

9. 部署与运维

在生产环境中部署和运维FastAPI应用程序是确保其稳定性和性能的关键步骤。

9.1 使用Docker进行部署

# 使用官方的 Python 镜像作为基础镜像
FROM python:3.10-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"]

解释：

• 使用官方的 Python 3.10 基础镜像。
• 设置工作目录为 /app。
• 复制 requirements.txt 并安装依赖。
• 复制项目文件。
• 使用 uvicorn 启动 FastAPI 应用。

9.2 使用GitHub Actions进行持续集成和部署

name: Deploy FastAPI App

on:
  push:
    branches:
      - main

jobs:
  build:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout code
      uses: actions/checkout@v2

    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: 3.10

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt

    - name: Build and push Docker image
      run: |
        docker build -t myfastapiapp .
        docker tag myfastapiapp mydockerhubuser/myfastapiapp:latest
        echo "${{ secrets.DOCKER_PASSWORD }}" | docker login -u "${{ secrets.DOCKER_USERNAME }}" --password-stdin
        docker push mydockerhubuser/myfastapiapp:latest

    - name: Deploy to server
      uses: appleboy/ssh-action@master
      with:
        host: ${{ secrets.SERVER_HOST }}
        username: ${{ secrets.SERVER_USER }}
        key: ${{ secrets.SERVER_SSH_KEY }}
        script: |
          docker pull mydockerhubuser/myfastapiapp:latest
          docker stop myfastapiapp_container || true
          docker rm myfastapiapp_container || true
          docker run -d --name myfastapiapp_container -p 8000:8000 mydockerhubuser/myfastapiapp:latest

解释：

• 定义了一个GitHub Actions工作流，在代码推送到 main 分支时触发。
• 检出代码，设置 Python 环境，安装依赖，构建并推送 Docker 镜像。
• 使用 SSH 登录服务器，拉取最新镜像，停止并删除旧容器，运行新容器。

9.3 配置日志

version: 1
formatters:
  access:
    "()": uvicorn.logging.AccessFormatter
    fmt: '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
handlers:
  console:
    class: logging.StreamHandler
    formatter: access
    stream: ext://sys.stdout
loggers:
  uvicorn:
    handlers: [console]
    level: INFO
    propagate: False
  uvicorn.error:
    level: INFO
    handlers: [console]
    propagate: False
  uvicorn.access:
    level: INFO
    handlers: [console]
    propagate: False

解释：

• 定义日志格式化器、处理器和记录器。
• AccessFormatter 格式化日志输出。
• StreamHandler 将日志输出到标准输出。
• 配置 uvicorn、uvicorn.error 和 uvicorn.access 记录器，设置日志级别为 INFO。

通过使用这些工具和配置，可以实现FastAPI应用程序的高效部署和稳定运维。

总结

本篇文章详细介绍了FastAPI的高级特性，包括依赖注入、中间件、事件处理、认证与授权、异步编程与性能优化、数据库集成与操作、测试与调试以及部署与运维。通过掌握这些高级特性和最佳实践，可以充分发挥FastAPI的强大功能，构建高性能、可维护的API服务。