graphene-federation

遵循Apollo Federation 规范的 Graphene 的联盟支持。

此存储库在很大程度上基于其分叉的存储库... 感谢 Preply 为建立基础。

警告：此版本与低于 v3 的 graphene 版本不兼容。如果您需要使用与 graphene v2 兼容的版本，我建议使用 graphene_federation 的 1.0.0 版本。

支持的功能

• sdl（字段上的 _service）：允许在联盟中添加模式（按原样）

Apollo 规范支持

• v1.0
• v2.0
• v2.1
• v2.2
• v2.3
• v2.4
• v2.5
• v2.6 STABLE_VERSION。Rover 开发版仅支持到 v2.6。
• v2.7 LATEST_VERSION

所有指令都可以通过 graphene-directives 轻松集成。现在，每个指令的值都由 graphene-directives 在运行时本身进行验证。

指令（v2.7）

directive @composeDirective(name: String!) repeatable on SCHEMA
directive @extends on OBJECT | INTERFACE
directive @external on OBJECT | FIELD_DEFINITION
directive @key(fields: FieldSet!, resolvable: Boolean = true) repeatable on OBJECT | INTERFACE
directive @inaccessible on
  | FIELD_DEFINITION
  | OBJECT
  | INTERFACE
  | UNION
  | ENUM
  | ENUM_VALUE
  | SCALAR
  | INPUT_OBJECT
  | INPUT_FIELD_DEFINITION
  | ARGUMENT_DEFINITION
directive @interfaceObject on OBJECT
directive @override(from: String!, label: String) on FIELD_DEFINITION
directive @provides(fields: FieldSet!) on FIELD_DEFINITION
directive @requires(fields: FieldSet!) on FIELD_DEFINITION
directive @shareable repeatable on FIELD_DEFINITION | OBJECT
directive @tag(name: String!) repeatable on
  | FIELD_DEFINITION
  | INTERFACE
  | OBJECT
  | UNION
  | ARGUMENT_DEFINITION
  | SCALAR
  | ENUM
  | ENUM_VALUE
  | INPUT_OBJECT
  | INPUT_FIELD_DEFINITION
directive @authenticated on
    FIELD_DEFINITION
  | OBJECT
  | INTERFACE
  | SCALAR
  | ENUM
directive @requiresScopes(scopes: [[federation__Scope!]!]!) on
    FIELD_DEFINITION
  | OBJECT
  | INTERFACE
  | SCALAR
  | ENUM
directive @policy(policies: [[federation__Policy!]!]!) on
  | FIELD_DEFINITION
  | OBJECT
  | INTERFACE
  | SCALAR
  | ENUM
scalar federation__Policy
scalar federation__Scope
scalar FieldSet

在官方文档中了解指令的更多信息

每个用 @key 或 @extends 装饰的类型都被添加到 _Entity 联合中。对于每个是实体的类型，可以定义一个 __resolve_reference 方法。注意，由于双下划线表示法在 Python 中可能导致模型继承问题，因此此解析方法也可以命名为 _resolve_reference（如果两者都声明，则 __resolve_reference 方法将具有优先权）。

每当请求实体作为满足查询计划的一部分时，都会调用此方法。如果没有显式定义，则使用默认解析器。默认解析器仅创建类型实例，其 kwargs 为传递的字段集，有关更多详细信息，请参阅 entity.get_entity_query

• 如果您需要在传递给字段解析器之前提取对象，则应定义 __resolve_reference（例如：FileNode）
• 如果字段解析器只需要在字段集中传递的数据，则不应定义 __resolve_reference（例如：FunnyText）。有关更多信息，请参阅官方文档。

示例

以下是基于Apollo Federation 介绍示例的实现示例。它通过三个服务（账户、产品、评论）实现了基本电子商务应用的联盟模式。

账户

首先添加一个账户服务，公开一个 User 类型，然后可以通过其 id 字段在其他服务中进行引用

from graphene import Field, Int, ObjectType, String

from graphene_federation import LATEST_VERSION, build_schema, key


@key("id")
class User(ObjectType):
    id = Int(required=True)
    username = String(required=True)

    def __resolve_reference(self, info, **kwargs):
        """
        Here we resolve the reference of the user entity referenced by its `id` field.
        """
        return User(id=self.id, email=f"user_{self.id}@mail.com")


class Query(ObjectType):
    me = Field(User)


schema = build_schema(query=Query, federation_version=LATEST_VERSION)

产品

产品服务公开一个 Product 类型，可以通过 upc 字段在其他服务中使用

from graphene import Argument, Int, List, ObjectType, String

from graphene_federation import LATEST_VERSION, build_schema, key


@key("upc")
class Product(ObjectType):
    upc = String(required=True)
    name = String(required=True)
    price = Int()

    def __resolve_reference(self, info, **kwargs):
        """
        Here we resolve the reference of the product entity referenced by its `upc` field.
        """
        return Product(upc=self.upc, name=f"product {self.upc}")


class Query(ObjectType):
    topProducts = List(Product, first=Argument(Int, default_value=5))


schema = build_schema(query=Query, federation_version=LATEST_VERSION)

评论

评论服务公开一个 Review 类型，该类型与 User 和 Product 类型相关联。它还具有提供 User 的用户名的功能。此外，它还向在其它服务中定义的 User/Product 类型添加了获取其评论的能力。

from graphene import Field, Int, List, ObjectType, String

from graphene_federation import LATEST_VERSION, build_schema, external, key, provides


@key("id")
class User(ObjectType):
    id = external(Int(required=True))
    reviews = List(lambda: Review)

    def resolve_reviews(self, info, *args, **kwargs):
        """
        Get all the reviews of a given user. (not implemented here)
        """
        return []


@key("upc")
class Product(ObjectType):
    upc = external(String(required=True))
    reviews = List(lambda: Review)


class Review(ObjectType):
    body = String()
    author = provides(Field(User), fields="username")
    product = Field(Product)


class Query(ObjectType):
    review = Field(Review)


schema = build_schema(query=Query, federation_version=LATEST_VERSION)

联盟

请注意，每个服务的模式声明都是有效的 GraphQL 模式（它仅添加了 _Entity 和 _Service 类型）。检查装饰器是否正确设置的最佳方法是请求服务的 SDL。

from graphql import graphql

query = """
query {
    _service {
        sdl
    }
}
"""

result = graphql(schema, query)
print(result.data["_service"]["sdl"])

然后可以在联盟模式中使用这些。

您可以在单元/集成测试中找到更多示例，以及示例文件夹。

还有关于与Mongoengine集成的示例。

其他注意事项

build_schema新参数

• schema_directives (Collection[SchemaDirective])：可以在DIRECTIVE_LOCATION.SCHEMA定义具有其参数值的指令。
• include_graphql_spec_directives (bool)：包含由GraphQL规范定义的指令（@include、@skip、@deprecated、@specifiedBy）。
• federation_version (FederationVersion)：指定版本（默认为STABLE_VERSION）。

指令附加参数

• federation_version：(FederationVersion = LATEST_VERSION)：您可以使用此功能从特定联盟版本中获取指令。

注意：在build_schema中，federation_version具有更高的优先级。如果您选择的指令不兼容，将引发错误。

自定义指令

您可以按照以下方式定义自定义指令

from graphene import Field, ObjectType, String
from graphql import GraphQLArgument, GraphQLInt, GraphQLNonNull

from graphene_federation import ComposableDirective, DirectiveLocation, LATEST_VERSION
from graphene_federation import build_schema

CacheDirective = ComposableDirective(
    name="cache",
    locations=[DirectiveLocation.FIELD_DEFINITION, DirectiveLocation.OBJECT],
    args={
        "maxAge": GraphQLArgument(
            GraphQLNonNull(GraphQLInt), description="Specifies the maximum age for cache in seconds."
        ),
    },
    description="Caching directive to control cache behavior.",
    spec_url="https://specs.example.dev/directives/v1.0",
)

cache = CacheDirective.decorator()


@cache(max_age=20)
class Review(ObjectType):
    body = cache(field=String(), max_age=100)


class Query(ObjectType):
    review = Field(Review)


schema = build_schema(
    query=Query,
    directives=(CacheDirective,),
    federation_version=LATEST_VERSION ,
)

这将自动将@link和@composeDirective添加到模式中

extend schema
	@link(url: "https://specs.apollo.dev/federation/v2.6", import: ["@composeDirective"])
	@link(url: "https://specs.example.dev/directives/v1.0", import: ["@cache"])
	@composeDirective(name: "@cache")

"""Caching directive to control cache behavior."""
directive @cache(
  """Specifies the maximum age for cache in seconds."""
  maxAge: Int!
) on FIELD_DEFINITION | OBJECT

type Query {
  review: Review
  _service: _Service!
}

type Review  @cache(maxAge: 20) {
  body: String @cache(maxAge: 100)
}

如果您希望手动添加schema_directives @link @composeDirective。您可以传递add_to_schema_directives为False

from graphene import Field, ObjectType, String
from graphql import GraphQLArgument, GraphQLInt, GraphQLNonNull

from graphene_federation import (ComposableDirective, DirectiveLocation, LATEST_VERSION, build_schema,
                                 compose_directive, link_directive)

CacheDirective = ComposableDirective(
    name="cache",
    locations=[DirectiveLocation.FIELD_DEFINITION, DirectiveLocation.OBJECT],
    args={
        "maxAge": GraphQLArgument(
            GraphQLNonNull(GraphQLInt), description="Specifies the maximum age for cache in seconds."
        ),
    },
    description="Caching directive to control cache behavior.",
    add_to_schema_directives=False
)

cache = CacheDirective.decorator()


@cache(max_age=20)
class Review(ObjectType):
    body = cache(field=String(), max_age=100)


class Query(ObjectType):
    review = Field(Review)


schema = build_schema(
    query=Query,
    directives=(CacheDirective,),
    schema_directives=(
        link_directive(url="https://specs.example.dev/directives/v1.0", import_=['@cache']),
        compose_directive(name='@cache'),
    ),
    federation_version=LATEST_VERSION,
)

自定义字段名称

在具有自定义名称的字段上使用装饰器时

情况1（auto_camelcase=False）

@key("identifier")
@key("validEmail")
class User(ObjectType):
    identifier = ID()
    email = String(name="validEmail")

class Query(ObjectType):
    user = Field(User)

schema = build_schema(query=Query, federation_version=LATEST_VERSION, auto_camelcase=False) # Disable auto_camelcase

这可以正确工作。默认情况下，如果auto_camelcase设置为False，则@key、@requires和@provides的fields不会转换为驼峰式。

情况2（auto_camelcase=True）

@key("identifier")
@key("valid_email")
class User(ObjectType):
    identifier = ID()
    email = String(name="valid_email")

class Query(ObjectType):
    user = Field(User)

schema = build_schema(query=Query, federation_version=LATEST_VERSION) # auto_camelcase Enabled

这将引发错误@key, field "validEmail" does not exist on type "User"。因为装饰器自动将键的field值转换为驼峰式，因为模式具有auto_camelcase=True（默认）。

要修复此问题，请在@key、@requires或@provides参数中传递auto_case=False。

@key("identifier")
@key("valid_email", auto_case=False)
class User(ObjectType):
    identifier = ID()
    email = String(name="valid_email")

class Query(ObjectType):
    user = Field(User)

schema = build_schema(query=Query, federation_version=LATEST_VERSION) # auto_camelcase=True

已知问题

• 在Federation v2.6中使用@composeDirective与@link显示rover中的错误，rover cli最多支持到v2.5，截至2024年1月16日。

贡献

• 您可以通过运行make tests来运行单元测试。
• 您可以通过运行make integration-build && make integration-test来运行集成测试。
• 您可以通过运行make dev-setup来获取开发环境（在Docker容器中）。
• 您应该使用black来格式化您的代码。

在GitHub上推送时，Travis CI会自动运行测试。