第六章：Python 实现

环境准备

安装依赖

# 安装 Graphene
pip install graphene flask-graphql

# 或者使用 Strawberry
pip install strawberry-graphql fastapi

# 或者使用 Ariadne
pip install ariadne uvicorn

Graphene 实现

定义 Schema

# schema.py
import graphene
from graphene import ObjectType, String, Int, ID, List, Schema

# 定义类型
class User(ObjectType):
    id = ID(required=True)
    name = String(required=True)
    email = String(required=True)
    age = Int()

# 定义查询
class Query(ObjectType):
    user = graphene.Field(User, id=ID(required=True))
    users = List(User)

    def resolve_user(self, info, id):
        # 从数据库获取用户
        return get_user_by_id(id)

    def resolve_users(self, info):
        # 获取所有用户
        return get_all_users()

# 定义变更
class CreateUser(graphene.Mutation):
    class Arguments:
        name = String(required=True)
        email = String(required=True)
        age = Int()

    user = graphene.Field(User)

    def mutate(self, info, name, email, age=None):
        user = create_user(name, email, age)
        return CreateUser(user=user)

class Mutation(ObjectType):
    create_user = CreateUser.Field()

# 创建 Schema
schema = Schema(query=Query, mutation=Mutation)

Flask 集成

# app.py
from flask import Flask
from flask_graphql import GraphQLView
from schema import schema

app = Flask(__name__)

# 添加 GraphQL 端点
app.add_url_rule(
    '/graphql',
    view_func=GraphQLView.as_view('graphql', schema=schema, graphiql=True)
)

if __name__ == '__main__':
    app.run(debug=True)

查询示例

# 执行查询
query = '''
    query {
        user(id: "1") {
            id
            name
            email
        }
    }
'''

result = schema.execute(query)
print(result.data)
# {'user': {'id': '1', 'name': 'John', 'email': 'john@example.com'}}

Strawberry 实现

定义 Schema

# schema.py
import strawberry
from typing import List, Optional

# 定义类型
@strawberry.type
class User:
    id: strawberry.ID
    name: str
    email: str
    age: Optional[int] = None

# 定义查询
@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: strawberry.ID) -> Optional[User]:
        return get_user_by_id(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, age: Optional[int] = None) -> User:
        return create_user(name, email, age)

# 创建 Schema
schema = strawberry.Schema(query=Query, mutation=Mutation)

FastAPI 集成

# app.py
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
from schema import schema

app = FastAPI()

# 添加 GraphQL 路由
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix='/graphql')

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

输入类型

# 定义输入类型
@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:
        return create_user(input.name, input.email, input.age)

订阅

import asyncio
from typing import AsyncGenerator

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def user_created(self) -> AsyncGenerator[User, None]:
        while True:
            # 等待新用户创建事件
            user = await wait_for_new_user()
            yield user

# 创建 Schema
schema = strawberry.Schema(
    query=Query,
    mutation=Mutation,
    subscription=Subscription
)

Ariadne 实现

定义 Schema

# schema.py
from ariadne import QueryType, MutationType, make_executable_schema

# 定义类型定义
type_defs = """
    type User {
        id: ID!
        name: String!
        email: String!
        age: Int
    }

    type Query {
        user(id: ID!): User
        users: [User!]!
    }

    type Mutation {
        createUser(name: String!, email: String!, age: Int): User!
    }
"""

# 定义查询
query = QueryType()

@query.field("user")
def resolve_user(_, info, id):
    return get_user_by_id(id)

@query.field("users")
def resolve_users(_, info):
    return get_all_users()

# 定义变更
mutation = MutationType()

@mutation.field("createUser")
def resolve_create_user(_, info, name, email, age=None):
    return create_user(name, email, age)

# 创建 Schema
schema = make_executable_schema(type_defs, query, mutation)

FastAPI 集成

# app.py
from fastapi import FastAPI
from ariadne.asgi import GraphQL
from schema import schema

app = FastAPI()

# 添加 GraphQL 端点
app.mount('/graphql', GraphQL(schema, debug=True))

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

数据库集成

SQLAlchemy 集成

# models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class UserModel(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)
    age = Column(Integer, nullable=True)

# schema.py
@strawberry.type
class User:
    id: strawberry.ID
    name: str
    email: str
    age: Optional[int] = None

    @classmethod
    def from_model(cls, model: UserModel) -> 'User':
        return cls(
            id=strawberry.ID(str(model.id)),
            name=model.name,
            email=model.email,
            age=model.age
        )

@strawberry.type
class Query:
    @strawberry.field
    def user(self, id: strawberry.ID) -> Optional[User]:
        model = session.query(UserModel).filter(UserModel.id == int(id)).first()
        return User.from_model(model) if model else None

小结

Python 实现要点：

• Graphene：定义 Schema、Flask 集成
• Strawberry：定义 Schema、FastAPI 集成、订阅
• Ariadne：定义 Schema、FastAPI 集成
• 数据库集成：SQLAlchemy

下一章我们将学习 Go 实现。