FastAPI Hello World —— 类型驱动的 API 开发
目标
创建一个最小的 FastAPI 应用,理解:
- FastAPI 的装饰器路由
- Pydantic 数据模型与类型验证
- 自动 OpenAPI 文档(Swagger / ReDoc)
- ASGI 异步支持
完整代码
# main.py
from fastapi import FastAPI, Path, Query, HTTPException
from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List
from enum import Enum
from datetime import datetime
# 创建 FastAPI 实例
app = FastAPI(
title="我的第一个 FastAPI",
description="FastAPI Hello World 示例",
version="1.0.0",
docs_url="/docs", # Swagger UI
redoc_url="/redoc", # ReDoc
)
# ---------- Pydantic 数据模型 ----------
class UserRole(str, Enum):
ADMIN = "admin"
USER = "user"
GUEST = "guest"
class User(BaseModel):
"""用户创建请求模型"""
username: str = Field(..., min_length=3, max_length=50, description="用户名")
email: EmailStr = Field(..., description="邮箱地址")
age: int = Field(ge=0, le=150, description="年龄")
role: UserRole = Field(default=UserRole.USER, description="用户角色")
tags: List[str] = Field(default=[], description="标签列表")
class UserResponse(BaseModel):
"""用户响应模型"""
id: int
username: str
email: str
role: UserRole
created_at: datetime
model_config = {"from_attributes": True} # 支持 ORM 对象
# ---------- 路由 ----------
@app.get("/")
async def root():
"""根路由 —— Hello World"""
return {
"message": "🚀 Hello, FastAPI!",
"docs": "/docs",
"redoc": "/redoc",
}
@app.get("/hello/{name}")
async def hello(
name: str = Path(..., min_length=1, max_length=50, description="用户名"),
greeting: str = Query("Hello", description="问候语"),
):
"""路径参数 + 查询参数"""
return {"message": f"{greeting}, {name}!"}
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: User):
"""创建用户 —— Pydantic 自动验证请求体"""
# 在实际应用中此处写数据库
if user.username == "admin":
raise HTTPException(status_code=400, detail="用户名 'admin' 已被保留")
return UserResponse(
id=42,
username=user.username,
email=user.email,
role=user.role,
created_at=datetime.now(),
)
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
user_id: int = Path(..., ge=1, description="用户 ID"),
):
"""获取用户详情"""
if user_id > 1000:
raise HTTPException(status_code=404, detail="用户不存在")
return UserResponse(
id=user_id,
username="demo_user",
email="demo@example.com",
role=UserRole.USER,
created_at=datetime.now(),
)
@app.get("/items/")
async def list_items(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
q: Optional[str] = Query(None, min_length=3),
):
"""分页查询 + 搜索"""
items = [{"id": i, "name": f"Item {i}"} for i in range(1, 31)]
if q:
items = [i for i in items if q.lower() in i["name"].lower()]
return {"items": items[skip : skip + limit], "total": len(items)}
# ---------- 自定义异常处理器 ----------
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
from fastapi.responses import JSONResponse
return JSONResponse(
status_code=exc.status_code,
content={
"error": exc.detail,
"path": str(request.url.path),
},
)
运行步骤
uvicorn main:app --reload
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
访问
测试 API
curl -X POST http:
-H "Content-Type: application/json" \
-d '{"username":"vibe","email":"vibe@test.com","age":25,"role":"admin","tags":["python","api"]}'
curl http:
curl "http://127.0.0.1:8000/items/?skip=0&limit=5&q=item"
curl -X POST http:
-H "Content-Type: application/json" \
-d '{"username":"abc","email":"bad","age":200,"role":"god"}'
关键要点
| 概念 |
说明 |
Body(...) / Path(...) / Query(...) |
参数来源标注 + 验证规则 |
... (Ellipsis) |
表示字段必填 |
Field(ge=0, le=150) |
数值范围约束 |
response_model |
响应模型,自动过滤多余字段、转换类型 |
Enum |
枚举类型,Swagger 自动显示下拉选择 |
HTTPException |
标准 HTTP 异常,返回结构化错误 |
async def |
异步视图,支持非阻塞 IO(数据库、HTTP 调用) |