Python API 开发：RESTful API 设计、FastAPI、OpenAPI 文档、认证鉴权

小飞兽 Python 8 次阅读 2026-07-24

Introduction

FastAPI 是现代 Python Web 框架，性能接近 Node.js，支持自动 OpenAPI 文档生成和类型提示。本文从 Hello World 开始，搭建完整的 RESTful API，包括路由、请求验证、Pydantic 模型、数据库集成、JWT 认证。

---

FastAPI 入门

pip install fastapi uvicorn pydantic python-jose passlib

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title='我的 API', version='1.0.0')

@app.get('/')
def read_root():
    return {'message': 'Hello, FastAPI!'}

@app.get('/items/{item_id}')
def read_item(item_id: int, q: str = None):
    return {'item_id': item_id, 'q': q}

启动：uvicorn main:app --reload --port 8000

自动文档：http://localhost:8000/docs（Swagger UI）或 http://localhost:8000/redoc（ReDoc）

Pydantic 请求/响应模型

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

class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr

class UserCreate(UserBase):
    password: str = Field(..., min_length=6)

class UserResponse(UserBase):
    id: int
    created_at: datetime
    class Config:
        from_attributes = True

class PostCreate(BaseModel):
    title: str = Field(..., max_length=200)
    content: str
    tags: List[str] = []

class PostResponse(PostCreate):
    id: int
    author_id: int
    views: int = 0
    created_at: datetime
    class Config:
        from_attributes = True

API 路由

from fastapi import APIRouter, HTTPException, Depends
from typing import List

router = APIRouter(prefix='/api/v1', tags=['文章'])

posts_db = []

@router.get('/posts', response_model=List[PostResponse])
def list_posts(skip: int = 0, limit: int = 10):
    return posts_db[skip: skip+limit]

@router.get('/posts/{post_id}', response_model=PostResponse)
def get_post(post_id: int):
    post = next((p for p in posts_db if p['id'] == post_id), None)
    if not post:
        raise HTTPException(status_code=404, detail='文章不存在')
    return post

@router.post('/posts', response_model=PostResponse, status_code=201)
def create_post(post: PostCreate):
    new_post = {
        'id': len(posts_db) + 1,
        **post.model_dump(),
        'author_id': 1,
        'views': 0,
        'created_at': datetime.now().isoformat()
    }
    posts_db.append(new_post)
    return new_post

@router.put('/posts/{post_id}', response_model=PostResponse)
def update_post(post_id: int, post: PostCreate):
    for p in posts_db:
        if p['id'] == post_id:
            p.update(post.model_dump())
            return p
    raise HTTPException(status_code=404, detail='文章不存在')

@router.delete('/posts/{post_id}', status_code=204)
def delete_post(post_id: int):
    global posts_db
    posts_db = [p for p in posts_db if p['id'] != post_id]
    return None

数据库集成

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from fastapi import Depends

DATABASE_URL = 'mysql+pymysql://user:pass@localhost/mydb'
engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@router.get('/users/{user_id}', response_model=UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail='用户不存在')
    return user

JWT 认证

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

SECRET_KEY = 'your-secret-key-here'
ALGORITHM = 'HS256'
ACCESS_TOKEN_EXPIRE_MINUTES = 60

pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/api/v1/auth/login')

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({'exp': expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@router.post('/auth/login')
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = db.query(User).filter(User.username == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail='用户名或密码错误')
    access_token = create_access_token(
        data={'sub': user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {'access_token': access_token, 'token_type': 'bearer'}

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get('sub')
    except JWTError:
        raise HTTPException(status_code=401, detail='Token 无效')
    user = db.query(User).filter(User.username == username).first()
    if not user:
        raise HTTPException(status_code=401, detail='用户不存在')
    return user

@router.get('/profile')
def get_profile(current_user: User = Depends(get_current_user)):
    return current_user

OpenAPI 文档定制

app = FastAPI(
    title='绍大技术网 API',
    description='提供文章、分类、用户相关接口',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redoc'
)

@router.get('/posts', summary='获取文章列表', description='支持分页和分类筛选')
def list_posts(page: int = 1, category: str = None):
    ...

常见问题

Q1: FastAPI 和 Flask 性能差距？FastAPI 基于 Starlette（异步），性能接近 Node.js；Flask 是同步框架，在 I/O 密集型场景 FastAPI 更快。

Q2: 如何处理文件上传？from fastapi import File, UploadFile; @app.post('/upload') async def upload(file: UploadFile = File(...))。

Q3: CORS 怎么配置？from fastapi.middleware.cors import CORSMiddleware; app.add_middleware(CORSMiddleware, allow_origins=['*'], allow_credentials=True, allow_methods=['*'], allow_headers=['*'])