`Python`的`GraphQL`后端：`Graphene`和`Ariadne`的`实现`与`对比`。

Python GraphQL 后端：Graphene 与 Ariadne 的实现与对比

大家好，今天我们要深入探讨如何在 Python 中构建 GraphQL 后端，重点比较两个流行的库：Graphene 和 Ariadne。我们将从基本概念入手，逐步分析它们的实现方式，并通过示例代码进行对比，最终帮助大家选择最适合自己项目的工具。

1. GraphQL 基础

在深入 Graphene 和 Ariadne 之前，我们先简单回顾一下 GraphQL 的核心概念。

• Schema (模式): 定义了 API 中可用的数据类型以及可以执行的操作（查询、变更和订阅）。
• Types (类型): 定义了数据的结构，包括字段及其类型。常见的类型包括 Scalar 类型 (Int, Float, String, Boolean, ID) 和 Object 类型。
• Resolvers (解析器): 负责从底层数据源获取数据，并将其返回给 GraphQL 查询引擎。每个字段都需要一个解析器。
• Query (查询): 用于读取数据的操作。
• Mutation (变更): 用于修改数据的操作。
• Subscription (订阅): 用于实时数据推送的操作。

2. Graphene：声明式 ORM 风格的 GraphQL 框架

Graphene 是一个强大的 Python GraphQL 框架，它提供了一种声明式的方式来定义 GraphQL schema，并且与许多流行的 ORM（如 SQLAlchemy 和 Django ORM）集成良好。

2.1 Graphene 的核心概念

• Object Types: 使用 graphene.ObjectType 定义 GraphQL 对象类型。
• Fields: 使用 graphene 模块中的字段类型 (例如 graphene.String, graphene.Int) 定义对象类型的字段。
• Resolvers: 通过在 Object Type 中定义名为 resolve_<field_name> 的方法来指定字段的解析器。
• Schema: 使用 graphene.Schema 创建 GraphQL schema。
• Query Type: 定义根查询类型，作为所有查询的入口点。
• Mutation Type: 定义根变更类型，作为所有变更的入口点。

2.2 Graphene 示例：简单的 To-Do List API

import graphene

# 定义 To-Do Item 类型
class Todo(graphene.ObjectType):
    id = graphene.ID()
    text = graphene.String()
    completed = graphene.Boolean()

# 定义查询类型
class Query(graphene.ObjectType):
    todos = graphene.List(Todo)

    def resolve_todos(root, info):
        # 模拟数据源
        return [
            Todo(id="1", text="Learn GraphQL", completed=True),
            Todo(id="2", text="Build a GraphQL API", completed=False),
        ]

# 定义变更类型
class CreateTodo(graphene.Mutation):
    class Arguments:
        text = graphene.String(required=True)

    todo = graphene.Field(Todo)

    def mutate(root, info, text):
        # 模拟创建 To-Do Item
        todo = Todo(id="3", text=text, completed=False)
        return CreateTodo(todo=todo)

class Mutation(graphene.ObjectType):
    create_todo = CreateTodo.Field()

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

# 执行查询
query = """
    query {
        todos {
            id
            text
            completed
        }
    }
"""
result = schema.execute(query)
print(result.data)

# 执行变更
mutation = """
    mutation {
        createTodo(text: "Buy groceries") {
            todo {
                id
                text
                completed
            }
        }
    }
"""
result = schema.execute(mutation)
print(result.data)

代码解释：

• 我们定义了一个 Todo 对象类型，包含 id, text, 和 completed 字段。
• Query 类型定义了一个 todos 字段，并使用 resolve_todos 方法作为解析器，模拟从数据源获取 To-Do Items。
• CreateTodo 变更类型定义了一个 text 参数，并使用 mutate 方法来模拟创建新的 To-Do Item。
• 我们创建了一个 Schema，并将 Query 和 Mutation 类型关联起来。
• 最后，我们使用 schema.execute 方法执行查询和变更，并打印结果。

2.3 Graphene 与 ORM 集成

Graphene 提供了与 SQLAlchemy 和 Django ORM 的集成，可以更方便地从数据库中获取数据。

2.3.1 Graphene 与 SQLAlchemy 集成

import graphene
from graphene import relay
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# SQLAlchemy Setup
DATABASE_URL = "sqlite:///:memory:"  # 使用内存数据库
engine = create_engine(DATABASE_URL)
db_session = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine))

Base = declarative_base()
Base.query = db_session.query_property()

class Department(Base):
    __tablename__ = 'department'
    id = Column(Integer, primary_key=True)
    name = Column(String)

class Employee(Base):
    __tablename__ = 'employee'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    department_id = Column(Integer)

Base.metadata.create_all(bind=engine)

# Graphene Setup
class DepartmentType(SQLAlchemyObjectType):
    class Meta:
        model = Department
        interfaces = (relay.Node, )

class EmployeeType(SQLAlchemyObjectType):
    class Meta:
        model = Employee
        interfaces = (relay.Node, )

class Query(graphene.ObjectType):
    node = relay.Node.Field()
    all_departments = SQLAlchemyConnectionField(DepartmentType.connection)
    all_employees = SQLAlchemyConnectionField(EmployeeType.connection)

schema = graphene.Schema(query=Query)

# 填充数据
department = Department(name='Engineering')
employee = Employee(name='John Doe', department_id=1)
db_session.add_all([department, employee])
db_session.commit()

# 执行查询
query = """
    query {
        allDepartments {
            edges {
                node {
                    id
                    name
                }
            }
        }
        allEmployees {
            edges {
                node {
                    id
                    name
                    departmentId
                }
            }
        }
    }
"""

result = schema.execute(query, context_value={'session': db_session})
print(result.data)

代码解释：

• 我们使用 SQLAlchemy 定义了 Department 和 Employee 模型。
• DepartmentType 和 EmployeeType 继承自 SQLAlchemyObjectType，并指定了对应的 SQLAlchemy 模型。
• SQLAlchemyConnectionField 提供了分页和排序功能。
• 查询通过 context_value 传递 SQLAlchemy session。

2.3.2 Graphene 与 Django ORM 集成

Graphene 与 Django 的集成更加紧密，可以直接使用 Django models 定义 GraphQL types。 需要安装 graphene-django 库。

# models.py (Django)
from django.db import models

class Category(models.Model):
    name = models.CharField(max_length=100)

    def __str__(self):
        return self.name

class Product(models.Model):
    name = models.CharField(max_length=100)
    price = models.DecimalField(max_digits=5, decimal_places=2)
    category = models.ForeignKey(Category, on_delete=models.CASCADE)

    def __str__(self):
        return self.name

# schema.py (Graphene)
import graphene
from graphene_django import DjangoObjectType
from .models import Category, Product

class CategoryType(DjangoObjectType):
    class Meta:
        model = Category
        fields = ("id", "name") # 可选，指定要暴露的字段

class ProductType(DjangoObjectType):
    class Meta:
        model = Product
        fields = ("id", "name", "price", "category")

class Query(graphene.ObjectType):
    all_categories = graphene.List(CategoryType)
    all_products = graphene.List(ProductType)
    category = graphene.Field(CategoryType, id=graphene.Int(required=True))

    def resolve_all_categories(root, info):
        return Category.objects.all()

    def resolve_all_products(root, info):
        return Product.objects.all()

    def resolve_category(root, info, id):
        try:
            return Category.objects.get(pk=id)
        except Category.DoesNotExist:
            return None

schema = graphene.Schema(query=Query)

代码解释：

• 我们使用 Django models 定义了 Category 和 Product 模型。
• CategoryType 和 ProductType 继承自 DjangoObjectType，并指定了对应的 Django model。
• 解析器直接使用 Django ORM 的查询方法 (例如 Category.objects.all(), Category.objects.get())。

2.4 Graphene 的优点和缺点

优点：

• 声明式风格： 使用 Object Types 和 Fields 声明式地定义 Schema，代码可读性高。
• ORM 集成： 与 SQLAlchemy 和 Django ORM 集成良好，可以方便地从数据库中获取数据。
• Relay 支持： 内置 Relay 支持，方便构建复杂的 GraphQL API。
• 类型安全： Python 的类型提示与 Graphene 结合，可以提供更好的类型安全。

缺点：

• 学习曲线较陡峭： 相比 Ariadne，Graphene 的学习曲线更陡峭，需要理解更多的概念。
• 代码冗余： 在某些情况下，需要编写较多的代码来定义 Schema。
• 灵活性较低： 声明式风格虽然提高了可读性，但也降低了灵活性。

3. Ariadne：代码优先、简单灵活的 GraphQL 框架

Ariadne 是一个轻量级的 Python GraphQL 框架，它采用代码优先的方式来定义 GraphQL schema，并且提供了更高的灵活性。

3.1 Ariadne 的核心概念

• ObjectType: 使用 ObjectType 类创建对象类型，并使用 field 装饰器将 Python 函数注册为字段的解析器。
• QueryType: 使用 QueryType 类创建根查询类型。
• MutationType: 使用 MutationType 类创建根变更类型。
• SubscriptionType: 使用 SubscriptionType 类创建根订阅类型。
• make_executable_schema: 使用 make_executable_schema 函数将 GraphQL schema 定义 (SDL) 和解析器绑定在一起。
• graphql_sync 和 graphql: 用于同步和异步执行GraphQL 查询

3.2 Ariadne 示例：简单的 To-Do List API

from ariadne import QueryType, MutationType, ObjectType, make_executable_schema, graphql_sync, ScalarType
from graphql import build_schema

# 定义 GraphQL schema (SDL)
type_defs = """
    type Todo {
        id: ID!
        text: String!
        completed: Boolean!
    }

    type Query {
        todos: [Todo!]!
    }

    type Mutation {
        createTodo(text: String!): Todo!
    }
"""

# 创建 QueryType 和 MutationType 实例
query = QueryType()
mutation = MutationType()
todo_type = ObjectType("Todo") # 为 Todo 对象创建类型，方便之后定义字段解析器

# 模拟数据源
todos_data = [
    {"id": "1", "text": "Learn GraphQL", "completed": True},
    {"id": "2", "text": "Build a GraphQL API", "completed": False},
]

# 定义查询解析器
@query.field("todos")
def resolve_todos(_, info):
    return todos_data

# 定义变更解析器
@mutation.field("createTodo")
def resolve_create_todo(_, info, text):
    new_todo = {"id": str(len(todos_data) + 1), "text": text, "completed": False}
    todos_data.append(new_todo)
    return new_todo

# 定义Todo类型的字段解析器（例如，如果需要自定义id的解析）
# @todo_type.field("id")
# def resolve_todo_id(obj, info):
#    return "custom_" + obj["id"]

# 创建 Schema
schema = make_executable_schema(type_defs, query, mutation, todo_type) # todo_type 必须传进去，否则找不到todo_type定义的解析器

# 执行查询
query_string = """
    query {
        todos {
            id
            text
            completed
        }
    }
"""
success, result = graphql_sync(
    schema,
    {"query": query_string},
    context_value=None, # 可用于传递请求上下文，例如用户认证信息
)

if success:
    print(result["data"])
else:
    print(result["errors"])

# 执行变更
mutation_string = """
    mutation {
        createTodo(text: "Buy groceries") {
            id
            text
            completed
        }
    }
"""
success, result = graphql_sync(
    schema,
    {"query": mutation_string},
    context_value=None,
)

if success:
    print(result["data"])
else:
    print(result["errors"])

代码解释：

• 我们首先定义了 GraphQL schema 使用 schema definition language (SDL)。
• QueryType 和 MutationType 用于定义根查询和变更类型。
• @query.field 和 @mutation.field 装饰器用于将 Python 函数注册为字段的解析器。
• make_executable_schema 函数将 GraphQL schema 定义和解析器绑定在一起。
• graphql_sync 函数用于执行 GraphQL 查询和变更。

3.3 Ariadne 与 ORM 集成

Ariadne 不像 Graphene 那样提供内置的 ORM 集成，但它可以与任何 ORM 框架配合使用。你需要手动编写解析器来从数据库中获取数据。

from ariadne import QueryType, make_executable_schema
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# SQLAlchemy Setup
DATABASE_URL = "sqlite:///:memory:"
engine = create_engine(DATABASE_URL)
Session = sessionmaker(bind=engine)
session = Session()

Base = declarative_base()

class Department(Base):
    __tablename__ = 'department'
    id = Column(Integer, primary_key=True)
    name = Column(String)

class Employee(Base):
    __tablename__ = 'employee'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    department_id = Column(Integer)

Base.metadata.create_all(engine)

# 填充数据
department = Department(name='Engineering')
employee = Employee(name='John Doe', department_id=1)
session.add_all([department, employee])
session.commit()

# Graphene Setup
type_defs = """
    type Department {
        id: ID!
        name: String!
    }

    type Employee {
        id: ID!
        name: String!
        departmentId: Int!
    }

    type Query {
        departments: [Department!]!
        employees: [Employee!]!
    }
"""

query = QueryType()

@query.field("departments")
def resolve_departments(_, info):
    return session.query(Department).all()

@query.field("employees")
def resolve_employees(_, info):
    return session.query(Employee).all()

schema = make_executable_schema(type_defs, query)

# 执行查询
query_string = """
    query {
        departments {
            id
            name
        }
        employees {
            id
            name
            departmentId
        }
    }
"""

from ariadne import graphql_sync
success, result = graphql_sync(
    schema,
    {"query": query_string},
    context_value={"session": session}  # 可选，传递session
)

if success:
    print(result["data"])
else:
    print(result["errors"])

代码解释：

• 我们使用 SQLAlchemy 定义了 Department 和 Employee 模型。
• 我们手动编写解析器 resolve_departments 和 resolve_employees 来从数据库中获取数据。
• 我们将 SQLAlchemy session 传递给 graphql_sync 函数的 context_value 参数。

3.4 Ariadne 的优点和缺点

优点：

• 简单易用： Ariadne 的 API 设计简洁明了，学习曲线较低。
• 灵活性高： 代码优先的方式提供了更高的灵活性，可以更方便地定制解析器。
• 轻量级： Ariadne 是一个轻量级的框架，依赖较少。
• 与任何 ORM 集成： 可以与任何 ORM 框架配合使用。

缺点：

• 需要手动编写更多代码： 相比 Graphene，需要手动编写更多的解析器代码。
• 类型安全较弱： 代码优先的方式可能导致类型安全较弱。
• 没有内置 Relay 支持： 需要手动实现 Relay 的相关功能。

4. Graphene 与 Ariadne 的对比

特性	Graphene	Ariadne
Schema 定义方式	声明式，ORM 风格	代码优先，SDL (Schema Definition Language)
ORM 集成	内置 SQLAlchemy 和 Django ORM 集成	需要手动实现
Relay 支持	内置 Relay 支持	需要手动实现
学习曲线	较陡峭	较低
灵活性	较低	较高
类型安全	较高，与 Python 类型提示结合	较弱
代码冗余	较高	较低
适用场景	需要与 ORM 集成，需要 Relay 支持的项目	追求灵活性和简单性的项目

5. 如何选择 Graphene 或 Ariadne？

选择 Graphene 还是 Ariadne 取决于你的具体项目需求。

• 选择 Graphene：
  • 你的项目需要与 SQLAlchemy 或 Django ORM 集成。
  • 你的项目需要 Relay 支持。
  • 你更喜欢声明式风格的 API 定义方式。
  • 你希望获得更好的类型安全。
• 选择 Ariadne：
  • 你的项目不需要与特定的 ORM 集成。
  • 你追求更高的灵活性和简单性。
  • 你更喜欢代码优先的方式定义 API。
  • 你希望使用更轻量级的框架。

6. 总结：选择合适的工具构建强大的 GraphQL API

Graphene 和 Ariadne 都是优秀的 Python GraphQL 框架，它们各有优缺点。 Graphene 提供了声明式风格和强大的 ORM 集成，而 Ariadne 则更加灵活和简单易用。 理解它们的差异，并根据你的项目需求选择最合适的工具，才能构建出高效、可维护的 GraphQL API。