本文将详细介绍如何使用 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.field
def hello(self) -> str:
"""测试接口"""
return "Hello World"
@strawberry.field
def me(self, info) -> Optional[UserRead]:
"""
获取当前用户信息
- 需要认证
- 返回 None 表示未登录
- 返回 UserRead 类型表示当前登录用户信息
"""
if not info.context.get("user"):
return None
return info.context["user"].to_read()
@strawberry.field
def 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 1
size = 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.field
def user_posts(self, username: str, page_input: Optional[PageInput] = None) -> Page[PostRead]:
"""
获取指定用户的文章列表
- 公开接口，无需认证
- 支持分页查询
- 返回带分页信息的文章列表
参数：
- username: 用户名
- page_input: 可选的分页参数
"""
# ... 实现类似 my_posts
@strawberry.type
class Mutation:
@strawberry.mutation
def 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.mutation
def register(self, user_data: UserCreate) -> RegisterResponse:
"""
用户注册
- 公开接口，无需认证
- 检查用户名和邮箱是否已存在
- 创建新用户
- 生成访问令牌
参数：
- user_data:
- username: 用户名
- password: 密码
- email: 邮箱
返回：
- token: 访问令牌
- user: 用户信息
"""
# ... 实现代码
@strawberry.mutation
def 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_delta
else:
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"] = user
finally:
db.close()
return context
3. 认证流程
- 用户登录：
mutation Login {
login(loginData: {
username: "admin",
password: "111111"
}) {
token {
accessToken
}
user {
id
username
email
}
}
}
-
服务器验证用户名密码，生成 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
mutation Login {
login(loginData: {
username: "admin",
password: "111111"
}) {
token {
accessToken
}
}
}
2. 使用 Token 访问私有接口
在 GraphQL Playground 中设置 HTTP Headers：
{
"Authorization": "Bearer your_token"
}
然后可以查询私有数据：
query MyPosts {
myPosts(pageInput: {
page: 1,
size: 10
}) {
items {
id
title
content
}
}
}
安全考虑
-
密码安全
- 使用 bcrypt 进行密码哈希
- 从不存储明文密码
-
Token 安全
- 使用 JWT 标准
- 设置合理的过期时间
- 使用安全的签名算法
-
数据访问控制
- 严格的权限检查
- 用户只能访问自己的数据
总结
本项目展示了如何使用现代化的技术栈构建一个安全的 GraphQL API：
- 使用 FastAPI 提供高性能的 Web 服务
- 使用 Strawberry 实现 GraphQL API
- 使用 SQLAlchemy 进行数据库操作
- 实现了完整的认证机制
- 遵循了最佳安全实践
当然图片上传一类的，还要跟以前一样写，但现在我们只写了一个/api接口就完成了项目所有接口。
