Python GraphQL API实战：构建灵活的数据查询接口-深圳市維司達科技有限公司

Python GraphQL API实战：构建灵活的数据查询接口

引言

在现代API开发中，GraphQL提供了一种灵活的数据查询方式。作为一名从Rust转向Python的后端开发者，我深刻体会到GraphQL在构建复杂API方面的优势。相比于REST API，GraphQL允许客户端精确获取所需数据，减少不必要的数据传输。

GraphQL 核心概念

什么是GraphQL

GraphQL是一种用于API的查询语言，具有以下特点：

灵活查询：客户端可以指定需要的数据结构
类型系统：强类型定义，提供编译时检查
单一端点：所有请求都通过一个端点处理
实时更新：支持订阅机制

架构设计

┌─────────────────────────────────────────────────────────────┐ │ GraphQL 客户端 │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ 查询定义 → 发送请求 → 解析响应 │ │ │ └─────────────────────────┬─────────────────────────┘ │ └─────────────────────────────┼─────────────────────────────┘ │ HTTP POST ▼ ┌─────────────────────────────────────────────────────────────┐ │ GraphQL 服务端 │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ 解析查询 → 验证类型 → 执行解析 → 返回结果 │ │ │ └─────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘

环境搭建与基础配置

安装依赖

pip install strawberry-graphql fastapi uvicorn

基本GraphQL服务

import strawberry from fastapi import FastAPI from strawberry.fastapi import GraphQLRouter @strawberry.type class User: id: int name: str email: str @strawberry.type class Query: @strawberry.field def user(self) -> User: return User(id=1, name="张三", email="zhangsan@example.com") schema = strawberry.Schema(query=Query) app = FastAPI() graphql_app = GraphQLRouter(schema) app.include_router(graphql_app, prefix="/graphql")

运行服务器

uvicorn main:app --reload

查询与变更

查询定义

@strawberry.type class Query: @strawberry.field def user(self, id: int) -> User: return get_user_from_database(id) @strawberry.field def users(self) -> list[User]: return get_all_users() @strawberry.type class Mutation: @strawberry.mutation def create_user(self, name: str, email: str) -> User: user = create_user_in_database(name, email) return user @strawberry.mutation def update_user(self, id: int, name: str) -> User: user = update_user_in_database(id, name) return user

复杂类型

@strawberry.type class Post: id: int title: str content: str author: User @strawberry.type class Query: @strawberry.field def post(self, id: int) -> Post: post = get_post(id) author = get_user(post.author_id) return Post( id=post.id, title=post.title, content=post.content, author=author )

高级特性实战

参数验证

from typing import Optional from strawberry import auto @strawberry.input class CreateUserInput: name: str email: str age: Optional[int] = None @strawberry.type class Mutation: @strawberry.mutation def create_user(self, input: CreateUserInput) -> User: if len(input.name) < 2: raise ValueError("Name must be at least 2 characters") user = User( id=generate_id(), name=input.name, email=input.email, age=input.age ) return user

分页查询

@strawberry.type class UserConnection: edges: list["UserEdge"] page_info: "PageInfo" @strawberry.type class UserEdge: node: User cursor: str @strawberry.type class PageInfo: has_next_page: bool has_previous_page: bool start_cursor: str end_cursor: str @strawberry.type class Query: @strawberry.field def users(self, first: int = 10, after: Optional[str] = None) -> UserConnection: users = get_users_with_pagination(first, after) edges = [ UserEdge(node=user, cursor=encode_cursor(user.id)) for user in users ] return UserConnection( edges=edges, page_info=PageInfo( has_next_page=has_more_users(), has_previous_page=after is not None, start_cursor=edges[0].cursor if edges else "", end_cursor=edges[-1].cursor if edges else "" ) )

订阅机制

import asyncio from strawberry.subscriptions import BaseSubscriptionResolver @strawberry.type class Subscription: @strawberry.subscription async def user_created(self) -> User: while True: user = await get_new_user() yield user await asyncio.sleep(1) schema = strawberry.Schema(query=Query, mutation=Mutation, subscription=Subscription)

实际业务场景

场景一：博客系统API

@strawberry.type class Comment: id: int content: str author: User post: Post @strawberry.type class Query: @strawberry.field def post(self, id: int) -> Post: post = db.query(PostModel).filter_by(id=id).first() author = db.query(UserModel).filter_by(id=post.author_id).first() comments = db.query(CommentModel).filter_by(post_id=id).all() return Post( id=post.id, title=post.title, content=post.content, author=User( id=author.id, name=author.name, email=author.email ), comments=[Comment( id=c.id, content=c.content, author=get_user(c.author_id), post=post ) for c in comments] )

场景二：购物车系统

@strawberry.type class Product: id: int name: str price: float stock: int @strawberry.type class CartItem: product: Product quantity: int @strawberry.type class Mutation: @strawberry.mutation def add_to_cart(self, user_id: int, product_id: int, quantity: int) -> list[CartItem]: cart = get_user_cart(user_id) product = get_product(product_id) if product.stock < quantity: raise ValueError("Insufficient stock") add_item_to_cart(cart, product, quantity) return get_cart_items(cart)

性能优化

数据加载器

from strawberry.dataloader import DataLoader async def batch_load_users(keys: list[int]) -> list[User]: users = db.query(UserModel).filter(UserModel.id.in_(keys)).all() user_map = {u.id: u for u in users} return [user_map.get(k) for k in keys] class MyGraphQLContext: def __init__(self): self.user_loader = DataLoader(batch_load_fn=batch_load_users) async def get_context() -> MyGraphQLContext: return MyGraphQLContext() @strawberry.type class Post: id: int title: str author_id: int @strawberry.field async def author(self, info) -> User: return await info.context.user_loader.load(self.author_id)

查询缓存

from functools import lru_cache @lru_cache(maxsize=128) def get_cached_user(user_id: int) -> User: return get_user_from_database(user_id) @strawberry.type class Query: @strawberry.field def user(self, id: int) -> User: return get_cached_user(id)