本文將詳細介紹如何使用 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. 認證流程

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接口就完成了項目所有接口。