Python 课程19-FastAPI

前言

FastAPI 是一个用于构建 API 的现代化、快速的 Python Web 框架。它基于 Python 的 type hints 构建，能够自动生成 API 文档并提供出色的性能。FastAPI 的设计目标是简单易用、高性能和支持异步操作，因此它非常适合开发高并发的 Web 应用程序和 API。

FastAPI 的核心优势在于它的速度快（与 Node.js 和 Go 相媲美）、支持异步功能、自动生成交互式文档以及使用 Python 类型提示来进行输入验证。

FastAPI 基础
- 安装 FastAPI 和 Uvicorn
- 创建一个简单的 FastAPI 应用
- FastAPI 自动生成的 API 文档
路由与请求处理
- 定义 API 路由
- 处理路径参数和查询参数
- 处理请求体（JSON）
- 返回响应
请求验证与数据模型
- 使用 Pydantic 进行数据验证
- 请求体模型与模式校验
- 数据验证与类型提示
高级功能
- 依赖注入
- 中间件
- 异步处理与性能优化
FastAPI 与数据库集成
- 使用 SQLAlchemy 与 FastAPI 集成
- CRUD 操作示例
- 连接池与事务管理
FastAPI 部署
- 使用 Uvicorn 部署应用
- 部署到云服务器或 Docker

1. FastAPI 基础

安装 FastAPI 和 Uvicorn

首先，你需要安装 FastAPI 和 Uvicorn。Uvicorn 是 FastAPI 推荐的 ASGI 服务器，用于运行 FastAPI 应用。

pip install fastapi uvicorn

创建一个简单的 FastAPI 应用

FastAPI 应用的创建非常简单，代码结构直观。

创建一个最简单的 FastAPI 应用：

from fastapi import FastAPI

# 创建 FastAPI 实例
app = FastAPI()

# 定义一个根路由
@app.get("/")
async def read_root():
    return {"message": "Hello, FastAPI!"}

在该示例中，我们创建了一个 FastAPI 应用实例 app，并定义了一个根路由 "/"，返回一个简单的 JSON 响应。

运行 FastAPI 应用

使用 Uvicorn 运行 FastAPI 应用：

uvicorn main:app --reload

main:app：main 是你的 Python 文件名，app 是 FastAPI 应用的实例。
--reload：启用自动重载，修改代码后无需手动重启服务器，适合开发环境。

FastAPI 自动生成的 API 文档

FastAPI 会自动生成交互式 API 文档。启动应用后，你可以在浏览器中访问以下 URL 来查看文档：

Swagger UI：http://127.0.0.1:8000/docs
ReDoc 文档：http://127.0.0.1:8000/redoc

FastAPI 使用 OpenAPI 规范，因此可以轻松生成和使用标准化的 API 文档。

2. 路由与请求处理

定义 API 路由

FastAPI 提供了多种 HTTP 方法，如 GET、POST、PUT、DELETE 等，你可以通过装饰器定义不同的路由。

定义不同的 HTTP 方法：

@app.get("/items/")
async def read_items():
    return {"items": ["item1", "item2"]}

@app.post("/items/")
async def create_item(item: dict):
    return {"item_created": item}

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: dict):
    return {"item_id": item_id, "updated_item": item}

处理路径参数和查询参数

FastAPI 可以轻松处理路径参数和查询参数，并会自动进行类型检查。

路径参数：

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

查询参数：

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

在此示例中，item_id 是路径参数，skip 和 limit 是查询参数。FastAPI 会自动将查询参数解析为整数并提供默认值。

处理请求体（JSON）

FastAPI 能够自动将请求体中的 JSON 数据解析为 Python 数据类型，并支持类型验证。

接收 JSON 请求体：

from pydantic import BaseModel

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

@app.post("/items/")
async def create_item(item: Item):
    return {"item_name": item.name, "item_price": item.price}

在此示例中，我们定义了一个 Pydantic 数据模型 Item，FastAPI 自动将请求体转换为 Item 类型，并进行数据验证。

返回响应

FastAPI 默认返回 JSON 响应，但你也可以返回其他格式的响应，如 HTML、文件等。

返回自定义响应：

from fastapi.responses import JSONResponse

@app.get("/custom_response/")
async def custom_response():
    return JSONResponse(content={"message": "This is a custom response"}, status_code=200)

3. 请求验证与数据模型

使用 Pydantic 进行数据验证

FastAPI 的核心是 Pydantic，它通过类型提示提供强大的数据验证功能。你可以通过 Pydantic 模型验证请求数据，并返回结构化的响应。

Pydantic 数据模型：

from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: str
    age: int

FastAPI 会自动生成验证错误的响应，如果数据格式不正确，API 会返回错误。

请求体模型与模式校验

通过 Pydantic 模型，你可以轻松定义请求体的结构并进行字段验证。

使用 Pydantic 进行模式校验：

class Product(BaseModel):
    name: str
    price: float
    in_stock: bool

@app.post("/products/")
async def create_product(product: Product):
    return product

在此示例中，FastAPI 会根据 Product 模型自动验证请求体数据的格式和类型。如果传递的数据不符合预期，会自动返回 422 错误。

数据验证与类型提示

通过 Python 的类型提示（type hints），FastAPI 能够为请求参数、请求体和响应提供自动化的数据验证。

查询参数与类型验证：

@app.get("/search/")
async def search_items(query: str, max_results: int = 10):
    return {"query": query, "max_results": max_results}

如果请求参数未通过验证，FastAPI 会返回 422 错误，并包含详细的错误信息。

4. 高级功能

依赖注入

FastAPI 提供了强大的依赖注入系统，用于复用逻辑、共享资源（如数据库连接）、认证和授权等。

使用依赖注入：

from fastapi import Depends

def get_current_user():
    return {"username": "john_doe"}

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

在此示例中，我们定义了一个 get_current_user() 函数，并通过 Depends() 将其注入到 API 路由中。

中间件

FastAPI 支持中间件，用于在请求处理前后执行自定义逻辑，例如日志记录、CORS、身份验证等。

定义中间件：

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # 处理请求前的逻辑
        response = await call_next(request)
        # 处理响应后的逻辑
        response.headers["X-Custom-Header"] = "Custom Value"
        return response

app.add_middleware(CustomMiddleware)

异步处理与性能优化

FastAPI 是异步框架，支持 async 和 await，可以处理大量并发请求。使用异步代码能够显著提高应用的性能。

定义异步路由：

@app.get("/async_example/")
async def async_example():
    # 模拟耗时操作
    await asyncio.sleep(1)
    return {"message": "This is an asynchronous response"}

5. FastAPI 与数据库集成

使用 SQLAlchemy 与 FastAPI 集成

FastAPI 可以与 SQLAlchemy 无缝集成，实现数据库操作。我们可以通过依赖注入来管理数据库会话，并在请求中使用数据库。

安装依赖

首先，你需要安装 SQLAlchemy 和数据库驱动程序。例如，使用 PostgreSQL 需要安装 psycopg2：

pip install sqlalchemy psycopg2

配置数据库与 SQLAlchemy

接下来，设置数据库引擎和会话管理。

创建数据库引擎与会话：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库连接字符串
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"

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

# 创建数据库会话类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建基础模型类
Base = declarative_base()

定义数据库模型

通过 SQLAlchemy 模型定义数据库表结构。

定义用户表模型：

from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)

创建数据库依赖

通过依赖注入方式将数据库会话注入到 FastAPI 路由中。以下代码中，get_db() 函数负责生成和关闭数据库会话。

依赖注入数据库会话：

from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session

# 获取数据库会话的依赖
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

定义 CRUD 操作

现在你可以在路由中定义 CRUD 操作。我们将通过数据库会话操作模型。

创建用户：

from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str
    email: str
    password: str

@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = User(name=user.name, email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

在这个示例中，我们定义了一个 UserCreate 数据模型，并创建了一个 /users/ 路由，用于创建用户。

查询用户：

这个路由通过用户 ID 查询用户数据，如果用户不存在，返回 404 错误。

连接池与事务管理

为了提高性能，可以使用连接池和事务管理。SQLAlchemy 提供了内置的连接池支持，FastAPI 通过 SessionLocal 处理会话和事务。

处理事务：

SQLAlchemy 的会话通过 commit() 提交事务，通过 rollback() 回滚事务。FastAPI 中，事务自动提交，但你也可以手动控制。

示例：回滚事务：

@app.post("/users/")
def create_user_with_transaction(user: UserCreate, db: Session = Depends(get_db)):
    try:
        db_user = User(name=user.name, email=user.email, hashed_password=user.password)
        db.add(db_user)
        db.commit()  # 提交事务
        db.refresh(db_user)
        return db_user
    except Exception:
        db.rollback()  # 如果出现异常，回滚事务
        raise HTTPException(status_code=400, detail="Error creating user")

6. FastAPI 部署

使用 Uvicorn 部署应用

FastAPI 应用使用 Uvicorn 部署。你可以通过 Uvicorn 的命令行工具或在 Python 中启动服务。

通过命令行部署：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

main:app：main 是你的应用文件，app 是 FastAPI 实例。
--host 0.0.0.0：绑定所有网络接口，适合在服务器中部署。
--reload：启用自动重载，适用于开发环境。

部署到 Docker

FastAPI 也可以通过 Docker 容器化部署，确保跨平台兼容性和部署一致性。

Dockerfile 示例：

# 使用官方的 Python 镜像
FROM python:3.9

# 设置工作目录
WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install -r requirements.txt

# 复制应用代码
COPY . .

# 启动应用
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

创建 Docker 镜像并运行容器：

docker build -t fastapi-app .
docker run -d -p 8000:8000 fastapi-app

通过 Docker，可以轻松将 FastAPI 应用打包，并在不同环境中运行。

部署到云服务器

FastAPI 可以轻松部署到 AWS、Google Cloud、Azure 或 DigitalOcean 等云服务器上。你可以通过以下步骤进行部署：

设置云服务器：通过 AWS EC2、Google Cloud Compute Engine 或其他云服务商启动虚拟机。
安装依赖：在服务器上安装 Python 和 FastAPI 所需的依赖。
配置防火墙和网络：开放端口 8000，允许外部访问 FastAPI 应用。
使用 Uvicorn 或 Gunicorn 运行：运行应用并确保其在后台持续运行（可以使用 supervisor 或 systemd 管理进程）。