FastAPI + GraphQL + SQLAlchemy 实现博客系统

本文将详细介绍如何使用 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: 可选的分页参数"""# ... 实现类似 [e-mail]
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. 认证流程

1. 用户登录：

mutation Login {login(loginData: {username: "admin",password: "111111"}) {token {accessToken}user {idusernameemail}}
}

2. 服务器验证用户名密码，生成 JWT token

3. 后续请求中使用 token：

   • 在请求头中添加：Authorization: Bearer your_token
   • 中间件解析 token 并验证
   • 将用户信息添加到 GraphQL context

4. 在需要认证的操作中检查用户：

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 {idtitlecontent}}
}

安全考虑

1. 密码安全

   • 使用 bcrypt 进行密码哈希
   • 从不存储明文密码

2. Token 安全

   • 使用 JWT 标准
   • 设置合理的过期时间
   • 使用安全的签名算法

3. 数据访问控制

   • 严格的权限检查
   • 用户只能访问自己的数据

总结

本项目展示了如何使用现代化的技术栈构建一个安全的 GraphQL API：

1. 使用 FastAPI 提供高性能的 Web 服务
2. 使用 Strawberry 实现 GraphQL API
3. 使用 SQLAlchemy 进行数据库操作
4. 实现了完整的认证机制
5. 遵循了最佳安全实践

当然图片上传一类的，还要跟以前一样写，但现在我们只写了一个/api接口就完成了项目所有接口。