本文将详细介绍如何使用 FastAPI、GraphQL（Strawberry）和 SQLAlchemy 实现一个带有认证功能的博客系统。
技术栈
- FastAPI：高性能的 Python Web 框架
- Strawberry：Python GraphQL 库
- SQLAlchemy：Python ORM 框架
- JWT：用于用户认证
系统架构
1. 数据模型（Models）
使用 SQLAlchemy 定义数据模型，以用户模型为例：
class UserModel(Base):"""SQLAlchemy model for the users table"""__tablename__ = "users"id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)username: Mapped[str] = mapped_column(String(50), unique=True, index=True)email: Mapped[str] = mapped_column(String(100), unique=True, index=True)hashed_password: Mapped[str] = mapped_column(String(200))nickname: Mapped[str] = mapped_column(String(50), nullable=True)is_active: Mapped[bool] = mapped_column(Boolean, default=True)is_admin: Mapped[bool] = mapped_column(Boolean, default=False)created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)# 关联关系posts = relationship("PostModel", back_populates="author")def verify_password(self, password: str) -> bool:"""验证密码"""return pwd_context.verify(password, self.hashed_password)
2. GraphQL Schema
使用 Strawberry 定义 GraphQL schema，包括查询和变更：
from models.types import (UserRead, # 用户信息读取类型UserCreate, # 用户创建输入类型LoginInput, # 登录输入类型LoginResponse, # 登录响应类型RegisterResponse, # 注册响应类型Token, # Token类型PostRead, # 文章读取类型PostCreate, # 文章创建输入类型PageInput, # 分页输入类型Page, # 分页响应类型PageInfo # 分页信息类型
)@strawberry.type
class Query:@strawberry.fielddef hello(self) -> str:"""测试接口"""return "Hello World"@strawberry.fielddef me(self, info) -> Optional[UserRead]:"""获取当前用户信息- 需要认证- 返回 None 表示未登录- 返回 UserRead 类型表示当前登录用户信息"""if not info.context.get("user"):return Nonereturn info.context["user"].to_read()@strawberry.fielddef my_posts(self, info, page_input: Optional[PageInput] = None) -> Page[PostRead]:"""获取当前用户的文章列表- 需要认证- 支持分页查询- 返回带分页信息的文章列表参数：- page_input: 可选的分页参数- page: 页码（默认1）- size: 每页大小（默认10）返回：- items: 文章列表- page_info: 分页信息- total: 总记录数- page: 当前页码- size: 每页大小- has_next: 是否有下一页- has_prev: 是否有上一页"""# 认证检查if not info.context.get("user"):raise ValueError("Not authenticated")# 数据库操作db = SessionLocal()try:# 设置分页参数page = page_input.page if page_input else 1size = page_input.size if page_input else 10# 查询总数total = db.query(func.count(PostModel.id)).filter(PostModel.author_id == info.context["user"].id).scalar()# 查询分页数据posts = (db.query(PostModel).options(joinedload(PostModel.author)) # 预加载作者信息.filter(PostModel.author_id == info.context["user"].id).order_by(PostModel.created_at.desc()) # 按创建时间倒序.offset((page - 1) * size).limit(size).all())# 构建分页信息page_info = PageInfo(total=total,page=page,size=size,has_next=total > page * size,has_prev=page > 1)return Page(items=[post.to_read() for post in posts],page_info=page_info)finally:db.close()@strawberry.fielddef user_posts(self, username: str, page_input: Optional[PageInput] = None) -> Page[PostRead]:"""获取指定用户的文章列表- 公开接口，无需认证- 支持分页查询- 返回带分页信息的文章列表参数：- username: 用户名- page_input: 可选的分页参数"""# ... 实现类似 my_posts@strawberry.type
class Mutation:@strawberry.mutationdef login(self, login_data: LoginInput) -> LoginResponse:"""用户登录- 公开接口，无需认证- 验证用户名密码- 生成访问令牌参数：- login_data:- username: 用户名- password: 密码返回：- token: 访问令牌- user: 用户信息"""db = SessionLocal()try:# 查找用户user = db.query(UserModel).filter(UserModel.username == login_data.username).first()# 验证密码if not user or not user.verify_password(login_data.password):raise ValueError("Incorrect username or password")# 生成访问令牌access_token = create_access_token(data={"sub": str(user.id)})token = Token(access_token=access_token)return LoginResponse(token=token, user=user.to_read())finally:db.close()@strawberry.mutationdef register(self, user_data: UserCreate) -> RegisterResponse:"""用户注册- 公开接口，无需认证- 检查用户名和邮箱是否已存在- 创建新用户- 生成访问令牌参数：- user_data:- username: 用户名- password: 密码- email: 邮箱返回：- token: 访问令牌- user: 用户信息"""# ... 实现代码@strawberry.mutationdef create_post(self, post_data: PostCreate, info) -> PostRead:"""创建文章- 需要认证- 创建新文章- 设置当前用户为作者参数：- post_data:- title: 标题- content: 内容返回：- 创建的文章信息"""# ... 实现代码schema = strawberry.Schema(query=Query, mutation=Mutation)
认证实现
1. JWT Token 生成
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:"""创建访问令牌"""to_encode = data.copy()if expires_delta:expire = datetime.utcnow() + expires_deltaelse:expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)to_encode.update({"exp": expire})encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)return encoded_jwt
2. 认证中间件
在 FastAPI 应用中实现认证中间件，用于解析和验证 token：
async def get_context(request: Request):"""GraphQL 上下文处理器，用于认证"""auth_header = request.headers.get("Authorization")context = {"user": None}if auth_header and auth_header.startswith("Bearer "):token = auth_header.split(" ")[1]token_data = verify_token(token)if token_data:db = SessionLocal()try:user = db.query(UserModel).filter(UserModel.id == int(token_data["sub"])).first()if user:context["user"] = userfinally:db.close()return context
3. 认证流程
- 用户登录：
graphql">mutation Login {login(loginData: {username: "admin",password: "111111"}) {token {accessToken}user {idusernameemail}}
}
-
服务器验证用户名密码，生成 JWT token
-
后续请求中使用 token：
- 在请求头中添加：
Authorization: Bearer your_token - 中间件解析 token 并验证
- 将用户信息添加到 GraphQL context
- 在请求头中添加：
-
在需要认证的操作中检查用户：
if not info.context.get("user"):raise ValueError("Not authenticated")
API 权限设计
1. 公开接口（无需认证）
hello: 测试接口login: 用户登录register: 用户注册userPosts: 获取指定用户的文章列表
2. 私有接口（需要认证）
me: 获取当前用户信息myPosts: 获取当前用户的文章列表createPost: 创建新文章
使用示例
1. 登录获取 Token
graphql">mutation Login {login(loginData: {username: "admin",password: "111111"}) {token {accessToken}}
}
2. 使用 Token 访问私有接口
在 GraphQL Playground 中设置 HTTP Headers：
{"Authorization": "Bearer your_token"
}
然后可以查询私有数据：
graphql">query MyPosts {myPosts(pageInput: {page: 1,size: 10}) {items {idtitlecontent}}
}
安全考虑
-
密码安全
- 使用 bcrypt 进行密码哈希
- 从不存储明文密码
-
Token 安全
- 使用 JWT 标准
- 设置合理的过期时间
- 使用安全的签名算法
-
数据访问控制
- 严格的权限检查
- 用户只能访问自己的数据
总结
本项目展示了如何使用现代化的技术栈构建一个安全的 GraphQL API：
- 使用 FastAPI 提供高性能的 Web 服务
- 使用 Strawberry 实现 GraphQL API
- 使用 SQLAlchemy 进行数据库操作
- 实现了完整的认证机制
- 遵循了最佳安全实践
当然图片上传一类的，还要跟以前一样写，但现在我们只写了一个/api接口就完成了项目所有接口。
