第一章练习答案
练习 1：创建基础 CRUD API
python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict
app = FastAPI()
class Item(BaseModel):
name: str
description: str = None
price: float
items: Dict[int, Item] = {}
item_id_counter = 1
# Create
@app.post("/items/", status_code=201)
async def create_item(item: Item):
global item_id_counter
items[item_id_counter] = item
item_id_counter += 1
return {"id": item_id_counter - 1, **item.dict()}
# Read
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return {"id": item_id, **items[item_id].dict()}
# Update
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
items[item_id] = item
return {"id": item_id, **item.dict()}
# Delete
@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
del items[item_id]
# List
@app.get("/items/")
async def list_items(skip: int = 0, limit: int = 10):
items_list = [{"id": k, **v.dict()} for k, v in items.items()]
return items_list[skip:skip + limit]练习 2：数据验证
python
from pydantic import BaseModel, Field, validator, EmailStr
class User(BaseModel):
username: str = Field(..., min_length=3, max_length=20, regex="^[a-zA-Z0-9_]+$")
email: EmailStr
age: int = Field(..., ge=18, le=120)
bio: str = Field(None, max_length=500)
@validator('username')
def username_alphanumeric(cls, v):
if not v.isalnum() and '_' not in v:
raise ValueError('用户名只能包含字母、数字和下划线')
return v
@validator('bio')
def bio_no_spam(cls, v):
if v and any(word in v.lower() for word in ['spam', '广告']):
raise ValueError('简介包含禁止内容')
return v
@app.post("/users/")
async def create_user(user: User):
return user测试：
bash
# 成功
curl -X POST http://localhost:8000/users/ \
-H "Content-Type: application/json" \
-d '{"username":"john_doe","email":"john@example.com","age":25}'
# 失败：年龄小于 18
curl -X POST http://localhost:8000/users/ \
-H "Content-Type: application/json" \
-d '{"username":"john_doe","email":"john@example.com","age":15}'练习 3：错误处理
python
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
app = FastAPI()
# 自定义异常
class ItemNotFoundException(Exception):
def __init__(self, item_id: int):
self.item_id = item_id
# 异常处理器
@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request: Request, exc: ItemNotFoundException):
return JSONResponse(
status_code=404,
content={
"error": "ItemNotFound",
"message": f"Item {exc.item_id} does not exist",
"suggestion": "Check the item ID and try again"
}
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=422,
content={
"error": "ValidationError",
"details": exc.errors(),
"body": exc.body
}
)
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id not in items:
raise ItemNotFoundException(item_id=item_id)
return items[item_id]练习 4：分页和搜索
python
from typing import Optional, List
class ItemResponse(BaseModel):
total: int
page: int
page_size: int
items: List[dict]
@app.get("/items/", response_model=ItemResponse)
async def search_items(
q: Optional[str] = None,
page: int = 1,
page_size: int = 10,
sort_by: str = "name"
):
# 搜索
filtered_items = items
if q:
filtered_items = {
k: v for k, v in items.items()
if q.lower() in v.name.lower()
}
# 排序
sorted_items = sorted(
filtered_items.items(),
key=lambda x: getattr(x[1], sort_by, x[1].name)
)
# 分页
total = len(sorted_items)
start = (page - 1) * page_size
end = start + page_size
paginated_items = [
{"id": k, **v.dict()}
for k, v in sorted_items[start:end]
]
return ItemResponse(
total=total,
page=page,
page_size=page_size,
items=paginated_items
)练习 5：日志和监控
python
import logging
import time
from fastapi import Request
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("app.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
@app.middleware("http")
async def log_requests(request: Request, call_next):
logger.info(f"Start: {request.method} {request.url}")
start_time = time.time()
try:
response = await call_next(request)
process_time = time.time() - start_time
logger.info(
f"End: {request.method} {request.url} - "
f"Status: {response.status_code} - "
f"Time: {process_time:.2f}s"
)
response.headers["X-Process-Time"] = str(process_time)
return response
except Exception as e:
logger.error(f"Error: {request.method} {request.url} - {str(e)}")
raise练习 6：测试
python
from fastapi.testclient import TestClient
import pytest
client = TestClient(app)
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test Item", "price": 10.5}
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "Test Item"
assert "id" in data
def test_get_item():
# 先创建
create_response = client.post(
"/items/",
json={"name": "Test", "price": 5.0}
)
item_id = create_response.json()["id"]
# 再获取
response = client.get(f"/items/{item_id}")
assert response.status_code == 200
assert response.json()["name"] == "Test"
def test_get_nonexistent_item():
response = client.get("/items/99999")
assert response.status_code == 404
def test_create_invalid_item():
response = client.post(
"/items/",
json={"name": "", "price": -10}
)
assert response.status_code == 422
@pytest.mark.parametrize("item_id,expected_status", [
(1, 200),
(99999, 404),
])
def test_get_item_parametrized(item_id, expected_status):
response = client.get(f"/items/{item_id}")
assert response.status_code == expected_status关键要点总结
- 类型提示驱动：让 FastAPI 自动处理验证
- Pydantic 模型：定义清晰的数据契约
- 异常处理：使用 HTTPException 和自定义处理器
- 日志记录：记录关键操作和错误
- 测试优先：使用 TestClient 编写测试
- 文档自动生成：访问 /docs 查看 API 文档
返回第一章概览