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.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. 认证流程

1. 用户登录：

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

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 {
      id
      title
      content
    }
  }
}

安全考虑

1. 密码安全

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

2. Token 安全

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

3. 数据访问控制

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

总结

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

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

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