FastAPI 是一款现代化的 Web 框架，可以非常方便地创建高性能 API。通过本流程记录一个带有权限验证和数据校验的项目，从初始化项目到最终部署的每一步操作。
1. 项目初始化与环境准备
- 新建项目文件夹，并进入项目目录。
- 创建虚拟环境（推荐使用
venv）并激活：
python -m venv venv
source venv/bin/activate # Linux/macOS
.\venv\Scripts\activate # Windows
安装必要依赖：包括 FastAPI、Uvicorn、SQLAlchemy、Pydantic 和 JWT 支持库（如 PyJWT）等。
pip install fastapi uvicorn sqlalchemy pydantic passlib python-dotenv
2. 项目结构与基本配置
创建项目结构，为各模块留出合适的目录：
project-root/
├── app/
│ ├── crud/ # 数据库操作（CRUD）函数
│ ├── database.py # 数据库连接配置
│ ├── dependencies.py # 依赖项管理（如认证依赖）
│ ├── main.py # FastAPI 主应用入口
│ ├── models/ # 数据库模型
│ ├── routers/ # 路由
│ ├── schemas/ # Pydantic 模型，用于数据校验
│ └── utils/ # 工具模块，如 JWT、响应格式等
└── .env # 环境变量配置文件
配置 .env 文件，用于存储敏感信息（如数据库连接、JWT 密钥等）：
DATABASE_URL=mysql://user:password@localhost:3306/dbname
JWT_SECRET_KEY=your_jwt_secret_key
JWT_ALGORITHM=HS256
JWT_EXPIRATION_HOURS=1
3. 配置数据库连接与模型创建
3.1 配置数据库
在 database.py 中，创建数据库连接会话。我们使用 SQLAlchemy 连接数据库：
# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
DATABASE_URL = os.getenv("DATABASE_URL")
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
3.2 创建模型
根据项目需求，定义用户、角色、权限等数据表的模型。这里我们以用户和角色模型为例：
# app/models/user.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from app.database import Base
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
password = Column(String(255), nullable=False)
role_id = Column(Integer, ForeignKey("role.id"))
role = relationship("Role", back_populates="users")
class Role(Base):
__tablename__ = "role"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), unique=True, nullable=False)
users = relationship("User", back_populates="role")
注意事项：
- 使用
ForeignKey设置关联关系，以便于进行权限和角色关联。 - 确保每个表有主键，且字段设置合适的
nullable选项。
4. 创建 CRUD 操作
在 crud 文件夹中，为每个模型编写相应的 CRUD 操作函数。这样可以分离业务逻辑与数据库操作，提升代码复用性和可读性。
# app/crud/user.py
from sqlalchemy.orm import Session
from app.models.user import User, Role
from app.schemas.user import UserCreate
def create_user(db: Session, user: UserCreate):
db_user = User(**user.dict())
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def get_user(db: Session, user_id: int):
return db.query(User).filter(User.id == user_id).first()
注意事项：
- 每个 CRUD 函数应返回查询的结果或执行的操作结果。
- 使用
session.commit()提交事务，session.refresh()更新本地实例。
5. 配置数据校验和 Pydantic 模型
在 schemas 文件夹中，创建 Pydantic 模型，用于请求和响应的数据校验：
# app/schemas/user.py
from pydantic import BaseModel
class UserBase(BaseModel):
username: str
class UserCreate(UserBase):
password: str
class UserResponse(UserBase):
id: int
注意事项：
- 将请求和响应的数据模型分离，确保接口的输出和输入规范。
6. 编写 JWT 认证逻辑
在 utils 文件夹下创建 jwt_helpers.py 文件，包含生成和验证 JWT Token 的方法。
# app/utils/jwt_helpers.py
import jwt
from datetime import datetime, timedelta
from fastapi import HTTPException, status
import os
SECRET_KEY = os.getenv("JWT_SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_HOURS = 1
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def verify_token(token: str):
try:
return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except jwt.ExpiredSignatureError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired"
)
except jwt.JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token"
)
7. 保护路由
在需要认证的路由中使用依赖项，确保只有持有有效 JWT Token 的用户可以访问。例如，将 Depends(jwt_auth) 加入到每个受保护的路由中。
# app/routers/product.py
from fastapi import APIRouter, Depends
from app.utils.jwt_helpers import verify_token
from app.schemas.product import ProductCreate
router = APIRouter()
@router.get("/products", dependencies=[Depends(verify_token)])
async def get_products():
return {"message": "This is a protected route"}
8. 添加登录和注册路由
在 auth.py 中创建用户登录和注册路由，成功登录后返回 JWT Token：
# app/routers/auth.py
from fastapi import APIRouter, HTTPException, Depends
from app.utils.jwt_helpers import create_access_token
from app.crud.user import get_user_by_username
from app.schemas.user import UserLogin
from app.dependencies import get_db
from sqlalchemy.orm import Session
router = APIRouter()
@router.post("/login")
def login(user: UserLogin, db: Session = Depends(get_db)):
db_user = get_user_by_username(db, user.username)
if not db_user or not verify_password(user.password, db_user.password):
raise HTTPException(status_code=400, detail="Invalid credentials")
access_token = create_access_token(data={"sub": db_user.username})
return {"access_token": access_token, "token_type": "bearer"}
9. 文档保护
将 Swagger 文档也设置为 JWT 保护，可以在文档访问之前要求登录。设置一个受保护的文档路由：
# app/main.py
from fastapi.openapi.docs import get_swagger_ui_html
from fastapi import FastAPI, Depends
from app.utils.jwt_helpers import verify_token
app = FastAPI(docs_url=None)
@app.get("/docs", dependencies=[Depends(verify_token)])
async def get_protected_docs():
return get_swagger_ui_html(openapi_url="/openapi.json", title="Protected API Docs")
10. 项目启动
启动项目的命令如下：
uvicorn app.main:app --host 0.0.0.0 --port 8001 --reload
通过本流程，我们完成了一个基础但安全性较高的 FastAPI 项目。总结下开发过程中的关键节点：
- 项目结构清晰，模块分明。
- JWT 认证用于保护敏感接口。
- Pydantic 模型用于请求和响应的严格数据校验。
- 密码加密和异常处理确保数据安全。
希望本教程能帮助你快速上手并构建安全的 API 服务。
