FastAPI 快速上手

概述

现在我们已经搭建好了开发环境，是时候创建第一个 FastAPI 应用了！本章将通过实际的代码示例，带您快速体验 FastAPI 的核心功能，包括路由创建、参数处理、数据验证和自动文档生成。

🚀 第一个 FastAPI 应用

最简单的 Hello World

创建 main.py 文件：

python

from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

# 创建路由端点
@app.get("/")
async def read_root():
    return {"Hello": "World"}

# 启动应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

运行应用：

bash

# 方法1: 直接运行
python main.py

# 方法2: 使用 uvicorn (推荐)
uvicorn main:app --reload

# 方法3: 指定参数
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

验证应用运行

bash

# 测试 API
curl http://localhost:8000/

# 期望输出: {"Hello":"World"}

访问自动生成的文档：

Swagger UI: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc

📊 添加更多端点

基础 CRUD 操作

python

from fastapi import FastAPI
from typing import Optional

app = FastAPI(
    title="My First FastAPI App",
    description="学习 FastAPI 的第一个应用",
    version="1.0.0"
)

# 模拟数据库
fake_items_db = [
    {"item_id": 1, "name": "Laptop", "price": 1000, "description": "高性能笔记本电脑"},
    {"item_id": 2, "name": "Mouse", "price": 25, "description": "无线鼠标"},
    {"item_id": 3, "name": "Keyboard", "price": 75, "description": "机械键盘"}
]

# GET: 获取所有物品
@app.get("/")
async def read_root():
    return {"message": "欢迎使用 FastAPI!", "items_count": len(fake_items_db)}

# GET: 获取所有物品（带查询参数）
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return fake_items_db[skip: skip + limit]

# GET: 获取单个物品
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Optional[str] = None):
    for item in fake_items_db:
        if item["item_id"] == item_id:
            if q:
                return {"item": item, "query": q}
            return {"item": item}
    return {"error": "Item not found"}

# POST: 创建新物品
@app.post("/items/")
async def create_item(name: str, price: float, description: Optional[str] = None):
    new_item_id = max([item["item_id"] for item in fake_items_db]) + 1
    new_item = {
        "item_id": new_item_id,
        "name": name,
        "price": price,
        "description": description or ""
    }
    fake_items_db.append(new_item)
    return {"message": "Item created", "item": new_item}

# PUT: 更新物品
@app.put("/items/{item_id}")
async def update_item(item_id: int, name: str, price: float, description: Optional[str] = None):
    for i, item in enumerate(fake_items_db):
        if item["item_id"] == item_id:
            fake_items_db[i] = {
                "item_id": item_id,
                "name": name,
                "price": price,
                "description": description or item["description"]
            }
            return {"message": "Item updated", "item": fake_items_db[i]}
    return {"error": "Item not found"}

# DELETE: 删除物品
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    for i, item in enumerate(fake_items_db):
        if item["item_id"] == item_id:
            deleted_item = fake_items_db.pop(i)
            return {"message": "Item deleted", "item": deleted_item}
    return {"error": "Item not found"}

🎯 使用 Pydantic 模型

定义数据模型

python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

app = FastAPI(title="商品管理 API", version="1.0.0")

# Pydantic 模型定义
class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100, description="物品名称")
    price: float = Field(..., gt=0, description="物品价格，必须大于0")
    description: Optional[str] = Field(None, max_length=500, description="物品描述")
    is_available: bool = Field(True, description="是否可用")

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    price: Optional[float] = Field(None, gt=0)
    description: Optional[str] = Field(None, max_length=500)
    is_available: Optional[bool] = None

class ItemResponse(ItemBase):
    item_id: int = Field(..., description="物品ID")
    created_at: datetime = Field(..., description="创建时间")
    updated_at: datetime = Field(..., description="更新时间")
    
    class Config:
        # 允许从 ORM 对象创建
        from_attributes = True

# 模拟数据库
fake_items_db: List[dict] = []
next_item_id = 1

# 辅助函数
def get_current_time():
    return datetime.now()

def find_item_by_id(item_id: int):
    for item in fake_items_db:
        if item["item_id"] == item_id:
            return item
    return None

# API 端点
@app.get("/", summary="根路径", description="返回 API 基本信息")
async def read_root():
    return {
        "message": "商品管理 API",
        "version": "1.0.0",
        "items_count": len(fake_items_db)
    }

@app.get("/items/", response_model=List[ItemResponse], summary="获取所有物品")
async def read_items(
    skip: int = Field(0, ge=0, description="跳过的记录数"),
    limit: int = Field(10, ge=1, le=100, description="返回的记录数")
):
    """
    获取物品列表
    
    - **skip**: 跳过的记录数（分页用）
    - **limit**: 返回的记录数（1-100）
    """
    return fake_items_db[skip: skip + limit]

@app.get("/items/{item_id}", response_model=ItemResponse, summary="获取单个物品")
async def read_item(item_id: int = Field(..., description="物品ID")):
    """
    根据 ID 获取物品详情
    
    - **item_id**: 物品的唯一标识符
    """
    item = find_item_by_id(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="物品未找到")
    return item

@app.post("/items/", response_model=ItemResponse, status_code=201, summary="创建新物品")
async def create_item(item: ItemCreate):
    """
    创建新物品
    
    - **name**: 物品名称（必需，1-100字符）
    - **price**: 物品价格（必需，大于0）
    - **description**: 物品描述（可选，最多500字符）
    - **is_available**: 是否可用（默认为 true）
    """
    global next_item_id
    
    current_time = get_current_time()
    new_item = {
        "item_id": next_item_id,
        "name": item.name,
        "price": item.price,
        "description": item.description,
        "is_available": item.is_available,
        "created_at": current_time,
        "updated_at": current_time
    }
    
    fake_items_db.append(new_item)
    next_item_id += 1
    
    return new_item

@app.put("/items/{item_id}", response_model=ItemResponse, summary="更新物品")
async def update_item(item_id: int, item: ItemUpdate):
    """
    更新现有物品
    
    - **item_id**: 要更新的物品ID
    - 只更新提供的字段，其他字段保持不变
    """
    existing_item = find_item_by_id(item_id)
    if existing_item is None:
        raise HTTPException(status_code=404, detail="物品未找到")
    
    # 更新字段
    update_data = item.dict(exclude_unset=True)
    for field, value in update_data.items():
        existing_item[field] = value
    
    existing_item["updated_at"] = get_current_time()
    
    return existing_item

@app.delete("/items/{item_id}", summary="删除物品")
async def delete_item(item_id: int):
    """
    删除物品
    
    - **item_id**: 要删除的物品ID
    """
    for i, item in enumerate(fake_items_db):
        if item["item_id"] == item_id:
            deleted_item = fake_items_db.pop(i)
            return {"message": "物品已删除", "item": deleted_item}
    
    raise HTTPException(status_code=404, detail="物品未找到")

🧪 API 测试

使用 curl 测试

bash

# 1. 获取所有物品
curl -X GET "http://localhost:8000/items/" -H "accept: application/json"

# 2. 创建新物品
curl -X POST "http://localhost:8000/items/" \
  -H "accept: application/json" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "MacBook Pro",
    "price": 2500.00,
    "description": "Apple MacBook Pro 16英寸",
    "is_available": true
  }'

# 3. 获取单个物品
curl -X GET "http://localhost:8000/items/1" -H "accept: application/json"

# 4. 更新物品
curl -X PUT "http://localhost:8000/items/1" \
  -H "accept: application/json" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "MacBook Pro M2",
    "price": 2800.00
  }'

# 5. 删除物品
curl -X DELETE "http://localhost:8000/items/1" -H "accept: application/json"

使用 Python requests 测试

创建 test_api.py：

python

import requests
import json

# API 基础 URL
BASE_URL = "http://localhost:8000"

def test_api():
    # 1. 测试根路径
    response = requests.get(f"{BASE_URL}/")
    print("根路径:", response.json())
    
    # 2. 创建物品
    new_item = {
        "name": "iPhone 15",
        "price": 999.00,
        "description": "最新的 iPhone",
        "is_available": True
    }
    
    response = requests.post(f"{BASE_URL}/items/", json=new_item)
    print("创建物品:", response.json())
    created_item = response.json()
    item_id = created_item["item_id"]
    
    # 3. 获取物品
    response = requests.get(f"{BASE_URL}/items/{item_id}")
    print("获取物品:", response.json())
    
    # 4. 更新物品
    update_data = {
        "price": 1099.00,
        "description": "iPhone 15 Pro Max"
    }
    
    response = requests.put(f"{BASE_URL}/items/{item_id}", json=update_data)
    print("更新物品:", response.json())
    
    # 5. 获取所有物品
    response = requests.get(f"{BASE_URL}/items/")
    print("所有物品:", response.json())
    
    # 6. 删除物品
    response = requests.delete(f"{BASE_URL}/items/{item_id}")
    print("删除物品:", response.json())

if __name__ == "__main__":
    test_api()

运行测试：

bash

python test_api.py

📚 自动文档探索

Swagger UI 功能

访问 http://localhost:8000/docs，您可以：

查看所有 API 端点
在线测试 API：点击 "Try it out" 按钮
查看请求/响应模型
下载 OpenAPI 规范

ReDoc 文档

访问 http://localhost:8000/redoc，获得更美观的文档展示。

OpenAPI JSON

访问 http://localhost:8000/openapi.json，获取原始的 OpenAPI 规范。

🔧 错误处理和状态码

添加自定义异常处理

python

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# 自定义异常类
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# 异常处理器
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "error": "Item not found",
            "message": f"Item with id {exc.item_id} does not exist",
            "type": "ItemNotFoundError"
        }
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "error": "Validation error",
            "message": "输入数据验证失败",
            "details": exc.errors()
        }
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": "HTTP error",
            "message": exc.detail,
            "status_code": exc.status_code
        }
    )

# 使用自定义异常的端点
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    item = find_item_by_id(item_id)
    if item is None:
        raise ItemNotFoundError(item_id)
    return item

🎛️ 配置和环境

环境变量配置

创建 .env 文件：

env

APP_NAME=FastAPI 学习应用
APP_VERSION=1.0.0
DEBUG=True
API_PREFIX=/api/v1

配置管理：

python

from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI App"
    app_version: str = "1.0.0"
    debug: bool = False
    api_prefix: str = "/api/v1"
    
    class Config:
        env_file = ".env"

settings = Settings()

app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    debug=settings.debug
)

# 使用 API 前缀
from fastapi import APIRouter

api_router = APIRouter(prefix=settings.api_prefix)

@api_router.get("/items/")
async def read_items():
    return {"items": []}

app.include_router(api_router)

🚀 生产就绪配置

完整的应用配置

python

from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 创建应用
app = FastAPI(
    title="生产级 FastAPI 应用",
    description="带有中间件和安全配置的 FastAPI 应用",
    version="1.0.0",
    docs_url="/api/docs",  # 自定义文档路径
    redoc_url="/api/redoc",
    openapi_url="/api/openapi.json"
)

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 添加可信主机中间件
app.add_middleware(
    TrustedHostMiddleware, 
    allowed_hosts=["localhost", "127.0.0.1", "yourdomain.com"]
)

# 自定义中间件：请求时间记录
@app.middleware("http")
async def add_process_time_header(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)
    logger.info(f"Request processed in {process_time:.4f} seconds")
    return response

# 健康检查端点
@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": time.time(),
        "version": "1.0.0"
    }

# 启动和关闭事件
@app.on_event("startup")
async def startup_event():
    logger.info("Application startup complete")

@app.on_event("shutdown")
async def shutdown_event():
    logger.info("Application shutdown complete")

📋 项目组织

建议的文件结构

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用入口
│   ├── config.py            # 配置管理
│   ├── models/              # Pydantic 模型
│   │   ├── __init__.py
│   │   └── item.py
│   ├── routers/             # 路由模块
│   │   ├── __init__.py
│   │   └── items.py
│   ├── services/            # 业务逻辑
│   │   ├── __init__.py
│   │   └── item_service.py
│   └── utils/               # 工具函数
│       ├── __init__.py
│       └── helpers.py
├── tests/                   # 测试代码
│   ├── __init__.py
│   └── test_main.py
├── requirements.txt         # 依赖
├── .env                     # 环境变量
└── README.md               # 项目说明

模块化的代码示例

app/models/item.py：

python

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    description: Optional[str] = Field(None, max_length=500)

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    price: Optional[float] = Field(None, gt=0)
    description: Optional[str] = Field(None, max_length=500)

class ItemResponse(ItemBase):
    item_id: int
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

app/routers/items.py：

python

from fastapi import APIRouter, HTTPException
from typing import List
from app.models.item import ItemCreate, ItemUpdate, ItemResponse

router = APIRouter(prefix="/items", tags=["items"])

@router.get("/", response_model=List[ItemResponse])
async def read_items():
    return []

@router.post("/", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate):
    # 业务逻辑
    pass

app/main.py：

python

from fastapi import FastAPI
from app.routers import items

app = FastAPI(title="模块化 FastAPI 应用")

# 包含路由
app.include_router(items.router, prefix="/api/v1")

@app.get("/")
async def root():
    return {"message": "模块化 FastAPI 应用"}

总结

本章我们通过实际代码体验了 FastAPI 的核心功能：

✅ 创建基础 API：理解 FastAPI 应用的基本结构
✅ 路由和端点：实现 CRUD 操作
✅ Pydantic 模型：数据验证和序列化
✅ 自动文档：Swagger UI 和 ReDoc
✅ 错误处理：异常处理和状态码
✅ 中间件配置：CORS、日志、安全
✅ 项目组织：模块化和最佳实践

现在您已经掌握了 FastAPI 的基础用法，接下来我们将深入学习路由系统和参数处理。

最佳实践

始终使用 Pydantic 模型进行数据验证
为每个端点编写清晰的文档字符串
使用适当的 HTTP 状态码
合理组织代码结构，保持模块化
利用 FastAPI 的自动文档功能进行 API 测试

在下一章中，我们将详细学习 FastAPI 的路由系统和 HTTP 方法处理。

FastAPI 快速上手 ​

概述 ​

🚀 第一个 FastAPI 应用 ​

最简单的 Hello World ​

验证应用运行 ​

📊 添加更多端点 ​

基础 CRUD 操作 ​

🎯 使用 Pydantic 模型 ​

定义数据模型 ​

🧪 API 测试 ​

使用 curl 测试 ​

使用 Python requests 测试 ​

📚 自动文档探索 ​

Swagger UI 功能 ​

ReDoc 文档 ​

OpenAPI JSON ​

🔧 错误处理和状态码 ​

添加自定义异常处理 ​

🎛️ 配置和环境 ​

环境变量配置 ​

🚀 生产就绪配置 ​

完整的应用配置 ​

📋 项目组织 ​

建议的文件结构 ​

模块化的代码示例 ​

总结 ​

FastAPI 快速上手

概述

🚀 第一个 FastAPI 应用

最简单的 Hello World

验证应用运行

📊 添加更多端点

基础 CRUD 操作

🎯 使用 Pydantic 模型

定义数据模型

🧪 API 测试

使用 curl 测试

使用 Python requests 测试

📚 自动文档探索

Swagger UI 功能

ReDoc 文档

OpenAPI JSON

🔧 错误处理和状态码

添加自定义异常处理

🎛️ 配置和环境

环境变量配置

🚀 生产就绪配置

完整的应用配置

📋 项目组织

建议的文件结构

模块化的代码示例

总结