使用 Python 和 FastAPI 构建高性能 REST API 完整指南

30 3 月, 2026 13点热度 0人点赞 0条评论

引言

在 2026 年的今天，API 开发已经成为现代软件开发的基石。无论是构建微服务、移动应用后端，还是为前端提供数据接口，选择一个合适的框架至关重要。FastAPI 作为 Python 生态中最现代的 Web 框架之一，以其卓越的性能、自动文档生成和类型安全特性，成为了开发者的首选。

本文将带你从零开始，学习如何使用 FastAPI 构建一个生产级别的 REST API，包括项目结构、数据库集成、认证授权、错误处理等核心内容。

为什么选择 FastAPI？

FastAPI 自 2018 年发布以来，迅速成为了 Python Web 框架中的明星。它的核心优势包括：

高性能：基于 Starlette 和 Pydantic，性能与 Node.js 和 Go 相当
自动文档：自动生成 OpenAPI 和 ReDoc 文档
类型安全：利用 Python 类型提示，减少 bug
异步支持：原生支持 async/await，处理高并发
易于学习：简洁的 API 设计，上手快速

项目初始化

首先，让我们创建项目结构并安装必要的依赖：


# 创建项目目录
mkdir fastapi-blog-api
cd fastapi-blog-api

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install fastapi uvicorn[standard] sqlalchemy pydantic python-jose[cryptography] passlib[bcrypt] python-multipart
pip install alembic  # 数据库迁移工具

项目结构建议：


fastapi-blog-api/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用入口
│   ├── config.py        # 配置管理
│   ├── database.py      # 数据库连接
│   ├── models.py        # 数据模型
│   ├── schemas.py       # Pydantic 模式
│   ├── crud.py          # 数据库操作
│   ├── auth.py          # 认证逻辑
│   └── routers/
│       ├── __init__.py
│       ├── posts.py     # 文章路由
│       ├── users.py     # 用户路由
│       └── auth.py      # 认证路由
├── alembic/             # 数据库迁移
├── requirements.txt
└── .env                 # 环境变量

核心代码实现

1. 配置管理（config.py）


from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    PROJECT_NAME: str = "博客 API"
    VERSION: str = "1.0.0"
    API_V1_STR: str = "/api/v1"
    
    # 数据库
    DATABASE_URL: str = "sqlite:///./blog.db"
    
    # JWT 配置
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    
    # CORS
    CORS_ORIGINS: list = ["http://localhost:3000"]
    
    class Config:
        env_file = ".env"

@lru_cache()
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

2. 数据库连接（database.py）


from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.config import settings

engine = create_engine(
    settings.DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite 需要
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

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

3. 数据模型（models.py）


from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Boolean
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True, nullable=False)
    content = Column(Text, nullable=False)
    summary = Column(String)
    published = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")

4. Pydantic 模式（schemas.py）


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

# 用户相关
class UserBase(BaseModel):
    email: EmailStr
    username: str

class UserCreate(UserBase):
    password: str

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

# 文章相关
class PostBase(BaseModel):
    title: str
    content: str
    summary: Optional[str] = None
    published: bool = False

class PostCreate(PostBase):
    pass

class PostUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None
    summary: Optional[str] = None
    published: Optional[bool] = None

class PostResponse(PostBase):
    id: int
    author_id: int
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

5. 认证模块（auth.py）


from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import User
from app.config import settings

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

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception
    return user

6. 文章路由（routers/posts.py）


from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import List
from app.database import get_db
from app.models import Post, User
from app.schemas import PostCreate, PostResponse, PostUpdate
from app.auth import get_current_user

router = APIRouter(prefix="/posts", tags=["文章"])

@router.get("/", response_model=List[PostResponse])
def read_posts(
    skip: int = 0,
    limit: int = 10,
    published: bool = Query(None),
    db: Session = Depends(get_db)
):
    query = db.query(Post)
    if published is not None:
        query = query.filter(Post.published == published)
    return query.offset(skip).limit(limit).all()

@router.get("/{post_id}", response_model=PostResponse)
def read_post(post_id: int, db: Session = Depends(get_db)):
    post = db.query(Post).filter(Post.id == post_id).first()
    if post is None:
        raise HTTPException(status_code=404, detail="文章未找到")
    return post

@router.post("/", response_model=PostResponse)
def create_post(
    post: PostCreate,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    db_post = Post(**post.dict(), author_id=current_user.id)
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

@router.put("/{post_id}", response_model=PostResponse)
def update_post(
    post_id: int,
    post_update: PostUpdate,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    db_post = db.query(Post).filter(Post.id == post_id).first()
    if db_post is None:
        raise HTTPException(status_code=404, detail="文章未找到")
    if db_post.author_id != current_user.id:
        raise HTTPException(status_code=403, detail="无权修改此文章")
    
    update_data = post_update.dict(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_post, key, value)
    
    db.commit()
    db.refresh(db_post)
    return db_post

@router.delete("/{post_id}")
def delete_post(
    post_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    db_post = db.query(Post).filter(Post.id == post_id).first()
    if db_post is None:
        raise HTTPException(status_code=404, detail="文章未找到")
    if db_post.author_id != current_user.id:
        raise HTTPException(status_code=403, detail="无权删除此文章")
    
    db.delete(db_post)
    db.commit()
    return {"message": "文章已删除"}

7. 应用入口（main.py）


from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.database import Base, engine
from app.routers import posts, users, auth

# 创建数据库表
Base.metadata.create_all(bind=engine)

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.VERSION,
    openapi_url=f"{settings.API_V1_STR}/openapi.json"
)

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(auth.router, prefix=settings.API_V1_STR)
app.include_router(users.router, prefix=settings.API_V1_STR)
app.include_router(posts.router, prefix=settings.API_V1_STR)

@app.get("/")
def root():
    return {"message": "欢迎使用博客 API", "docs": "/docs"}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

运行与测试

启动服务器：


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

访问以下地址：

API 文档：http://localhost:8000/docs（Swagger UI）
ReDoc 文档：http://localhost:8000/redoc
健康检查：http://localhost:8000/health

最佳实践建议

环境变量：敏感信息（如密钥、数据库密码）务必使用环境变量
日志记录：使用 logging 模块记录关键操作和错误
速率限制：使用 slowapi 等库防止 API 滥用
单元测试：使用 pytest 编写测试，保证代码质量
容器化：使用 Docker 打包，便于部署
CI/CD：配置自动化测试和部署流程

总结

通过本文，我们完成了从零开始构建一个完整的 FastAPI REST API 项目。我们涵盖了项目结构、数据库集成、JWT 认证、CRUD 操作等核心内容。FastAPI 凭借其出色的性能和开发体验，确实是构建现代 API 的优秀选择。

下一步，你可以考虑添加更多功能，如：

文件上传（文章配图）
评论系统
标签和分类
全文搜索
Redis 缓存
消息队列（Celery）处理异步任务

希望本教程对你有所帮助！如果你有任何问题，欢迎在评论区留言讨论。

喜欢这篇文章吗？欢迎订阅我的博客获取更多技术教程！

本作品采用知识共享署名 4.0 国际许可协议进行许可