Skip to content

Federation🔗

Undine provides support for Apollo Federation 2, letting you expose an Undine schema as a subgraph in a federated supergraph. Federation 2 lets a router combine multiple subgraphs into a single GraphQL API, so different teams can own different parts of the schema.

Setup🔗

You can build a Federation 2 subgraph-compliant schema by importing create_federation_schema instead of create_schema.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from undine import Entrypoint, Field, QueryType, RootType
from undine.federation import create_federation_schema

from .models import Task


class TaskType(QueryType[Task]):
    id = Field()
    name = Field()


class Query(RootType):
    task = Entrypoint(TaskType)


schema = create_federation_schema(query=Query)

Compared with create_schema, create_federation_schema additionally:

  1. Validates FEDERATION_VERSION is supported.
  2. Auto-generates a schema-level @link directive.
  3. Injects Query._service which returns the subgraph SDL for the router.
  4. Injects Query._entities which fetches entities by their representations.

To publish the subgraph SDL to a router or a schema registry, export it with python manage.py print_schema. See Schema export for details.

Directives🔗

LinkDirective🔗

Subgraph schemas opt in to Apollo Federation 2 by applying the @link directive to the schema type. Undine automatically compiles the correct directive for you in create_federation_schema based on the directives in use in your schema.

1
extend schema @link(url: "https://specs.apollo.dev/federation/v2.11", import: ["@key"])

The @link directive also accepts two optional arguments: as renames the linked spec's namespace prefix, so directives are addressed as @<as>__key instead of @key, and for declares the purpose of the link, either SECURITY or EXECUTION, which Apollo Router uses to decide how to compose the subgraph. See the spec for more information.

KeyDirective🔗

The @key directive marks a QueryType as an entity.

To use the directive, you must define a set of fields that a subgraph can use to uniquely identify any instance of the entity. These fields must be defined on the QueryType.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import KeyDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field()

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  name: String!
}

Set resolvable=False when this subgraph only references the entity but does not resolve it. The router uses the declaration to stitch results together, but never routes resolution requests here.

1
2
3
4
5
6
7
8
9
from undine import Field, QueryType
from undine.federation import KeyDirective

from .models import Task


@KeyDirective(fields="id", resolvable=False)
class TaskType(QueryType[Task]):
    id = Field()

1
2
3
type TaskType @key(fields: "id", resolvable: false) {
  id: Int!
}

ShareableDirective🔗

the @shareable directive indicates that a field, or all fields of a type, are allowed to be resolved by multiple subgraphs.

1
2
3
4
5
6
7
8
9
from undine import Field, QueryType
from undine.federation import ShareableDirective

from .models import Task


class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ ShareableDirective()

1
2
3
4
type TaskType {
  id: Int!
  name: String! @shareable
}

Note: if a field is included in an entity's @key directive, that field is automatically considered @shareable and the directive is not required in the corresponding subgraph(s).

ExternalDirective🔗

The @external directive marks a field as owned by another subgraph. This subgraph declares the field so it can be referenced (e.g. by @requires) but does not resolve it.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import ExternalDirective, KeyDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ ExternalDirective()

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  name: String! @external
}

RequiresDirective🔗

The @requires directive declares that resolving a field on this subgraph requires additional @external fields from the entity to be fetched first.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from undine import Field, QueryType
from undine.federation import ExternalDirective, KeyDirective, RequiresDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ ExternalDirective()
    display_name = Field() @ RequiresDirective(fields="name")

1
2
3
4
5
type TaskType @key(fields: "id") {
  id: Int!
  name: String! @external
  displayName: String! @requires(fields: "name")
}

ProvidesDirective🔗

The @provides directive tells the router that this subgraph can resolve the listed fields of a referenced entity, avoiding an extra round-trip to the owning subgraph.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import KeyDirective, ProvidesDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    project = Field() @ ProvidesDirective(fields="name")

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  project: ProjectType! @provides(fields: "name")
}

OverrideDirective🔗

The @override directive tells the router that this subgraph takes over resolution of a field previously owned by another subgraph. from_ names the subgraph being overridden.

Pass an optional label to opt into progressive override — the router uses the label to route a configurable fraction of traffic to this subgraph rather than switching over all at once.

Requires FEDERATION_VERSION "2.7" or higher for the label argument.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import KeyDirective, OverrideDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ OverrideDirective(from_="legacy")

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  name: String! @override(from: "legacy")
}

InaccessibleDirective🔗

The @inaccessible directive hides a schema element from the supergraph. The element remains defined in the subgraph but the router never exposes it to clients.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import InaccessibleDirective, KeyDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    internal_note = Field() @ InaccessibleDirective()

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  internalNote: String @inaccessible
}

TagDirective🔗

The @tag directive attaches arbitrary metadata strings to schema elements. Contract composition and other Apollo tooling read the tags for filtering. Repeatable.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import KeyDirective, TagDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ TagDirective(name="public") @ TagDirective(name="v2")

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  name: String! @tag(name: "public") @tag(name: "v2")
}

ComposeDirectiveDirective🔗

The @composeDirective directive tells the router to preserve a custom directive from this subgraph in the supergraph schema. Apply it at the schema level via schema_definition_directives.

Requires FEDERATION_VERSION "2.1" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from undine import Entrypoint, Field, QueryType, RootType
from undine.federation import ComposeDirectiveDirective, create_federation_schema

from .models import Task


class TaskType(QueryType[Task]):
    name = Field()


class Query(RootType):
    task = Entrypoint(TaskType)


schema = create_federation_schema(
    query=Query,
    schema_definition_directives=[ComposeDirectiveDirective(name="@custom")],
)

1
extend schema @composeDirective(name: "@custom")

InterfaceObjectDirective🔗

The @interfaceObject directive lets a subgraph contribute fields to an entity that is declared as an interface in another subgraph. Apply it to a QueryType alongside @key.

Requires FEDERATION_VERSION "2.3" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import InterfaceObjectDirective, KeyDirective

from .models import Task


@InterfaceObjectDirective()
@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()

1
2
3
type TaskType @interfaceObject @key(fields: "id") {
  id: Int!
}

AuthenticatedDirective🔗

The @authenticated directive marks a schema element as requiring an authenticated request. Enforcement happens at the router.

Requires FEDERATION_VERSION "2.5" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import AuthenticatedDirective, KeyDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ AuthenticatedDirective()

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  name: String! @authenticated
}

RequiresScopesDirective🔗

The @requiresScopes directive marks a schema element as requiring the caller to hold one of the listed scope sets. Each inner list is an "all-of" group; the outer list is "any-of".

Requires FEDERATION_VERSION "2.5" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import KeyDirective, RequiresScopesDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ RequiresScopesDirective(scopes=[["read:task"]])

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  name: String! @requiresScopes(scopes: [["read:task"]])
}

PolicyDirective🔗

The @policy directive marks a schema element as gated by named authorization policies. The list-of-lists shape matches @requiresScopes.

Requires FEDERATION_VERSION "2.6" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import KeyDirective, PolicyDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ PolicyDirective(policies=[["policy_a"], ["policy_b"]])

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  name: String! @policy(policies: [["policy_a"], ["policy_b"]])
}

ContextDirective🔗

The @context directive names a context that the annotated type contributes to. Arguments elsewhere resolve their value from that context via @fromContext. Repeatable.

Requires FEDERATION_VERSION "2.8" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import ContextDirective, KeyDirective

from .models import Task


@ContextDirective(name="workspace")
@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()

1
2
3
type TaskType @context(name: "workspace") @key(fields: "id") {
  id: Int!
}

FromContextDirective🔗

The @fromContext directive populates a directive argument from a named context declared via @context.

Requires FEDERATION_VERSION "2.8" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from graphql import DirectiveLocation, GraphQLNonNull, GraphQLString

from undine.directives import Directive, DirectiveArgument
from undine.federation import FromContextDirective


class ScopedDirective(
    Directive,
    locations=[DirectiveLocation.FIELD_DEFINITION],
    schema_name="scoped",
):
    workspace = DirectiveArgument(
        GraphQLNonNull(GraphQLString),
        directives=[FromContextDirective(field="$workspace { id }")],
    )

CostDirective🔗

The @cost directive attaches a demand-control weight to a schema element. The router's cost analyzer sums these to estimate the total cost of a query.

Requires FEDERATION_VERSION "2.9" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from undine import Field, QueryType
from undine.federation import CostDirective, KeyDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ CostDirective(weight=5)

1
2
3
4
type TaskType @key(fields: "id") {
  id: Int!
  name: String! @cost(weight: 5)
}

ListSizeDirective🔗

The @listSize directive tells the demand-control cost analyzer how large a list-returning field can grow. assumed_size is a fixed upper bound; slicing_arguments and sized_fields describe pagination-based bounds.

Requires FEDERATION_VERSION "2.9" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from undine import Field, QueryType
from undine.federation import KeyDirective, ListSizeDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    comments = Field() @ ListSizeDirective(
        assumed_size=100,
        slicing_arguments=["first", "last"],
        sized_fields=["edges"],
    )

1
2
3
4
5
6
7
8
type TaskType @key(fields: "id") {
  id: Int!
  comments: [String!]! @listSize(
    assumedSize: 100,
    sizedFields: ["edges"],
    slicingArguments: ["first", "last"]
  )
}

CacheTagDirective🔗

The @cacheTag directive attaches a cache-tag template to a schema element. The router uses the rendered tags to invalidate response cache entries. Repeatable, and applicable to both object types and individual fields.

Requires FEDERATION_VERSION "2.12" or higher.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from undine import Field, QueryType
from undine.federation import CacheTagDirective, KeyDirective

from .models import Task


@CacheTagDirective(format="task")
@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field() @ CacheTagDirective(format="task:{$response.id}:name")

1
2
3
4
type TaskType @cacheTag(format: "task") @key(fields: "id") {
  id: Int!
  name: String! @cacheTag(format: "task:{$response.id}:name")
}

Entities🔗

An entity is an object type marked with @key that can be fetched with one or more unique keys and resolve its fields from multiple data sources in a federated graph. The router fetches entities across subgraphs by sending representations (dicts like { "__typename": "TaskType", "id": 1 }) to your subgraph and expects the matching entities back.

Undine wires this up automatically in create_federation_schema whenever the schema contains at least one resolvable @key entity, adding an _entities field to the Query type that the router calls to fetch entities by their representations.

Reference resolver🔗

The reference resolver runs for each representation after the router dispatches by __typename. By default, Undine builds one for you, but the default resolver only supports a QueryType with a single resolvable @key whose fields is a single, unaliased token that maps to a declared Field. In any other case, you need define a custom __resolve_reference__ on the QueryType.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from typing import Any

from undine import Field, GQLInfo, QueryType
from undine.federation import KeyDirective

from .models import Task


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field()

    @classmethod
    def __resolve_reference__(cls, representation: dict[str, Any], info: GQLInfo) -> Task | None:
        return Task.objects.filter(pk=representation["id"]).first()

Returning None from __resolve_reference__ yields null for that entry. Raising propagates as a GraphQL error on that entry's response slot.

FederationType🔗

A FederationType contributes fields to an entity that is owned by another subgraph and is not backed by a Django model. You still mark it with @KeyDirective so the router knows how to identify entities of this type.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from undine.federation import FederationField, FederationType, KeyDirective

from .models import Task


@KeyDirective(fields="id")
class UserExtension(FederationType, schema_name="User"):
    id = FederationField(int)
    assigned_task_count = FederationField(int)

    @assigned_task_count.resolve
    def resolve_assigned_task_count(self, info) -> int:
        # `self.id` reads from the representation the router sent.
        return Task.objects.filter(assigned_to_id=self.id).count()

1
2
3
4
type User @key(fields: "id") {
  id: Int!
  assignedTaskCount: Int!
}

Note: schema_name must match the name to the shared entity name across subgraphs.

FederationField🔗

A FederationField is used to define a queryable value on a FederationType. It mirrors Field on a QueryType.

A FederationField resolves via one of three rules:

  1. Explicit resolver — @<field>.resolve was used, or the decorator form @FederationField(...) wrapped a function.
  2. Key field — the field's name appears in some @KeyDirective(fields=...). The router populates the attribute from the representation; the resolver is a plain attribute lookup on the instance.
  3. @external — the field carries @ExternalDirective(). Populated by another subgraph via @requires; again, a plain attribute lookup.

Anything else — a computed field with no resolver, no key membership, and no @external — is a definition error and raises an exception when the schema is built.

The following declares a computed overdue_task_count that depends on timezone from another subgraph:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from undine import GQLInfo
from undine.federation import ExternalDirective, FederationField, FederationType, KeyDirective, RequiresDirective

from .models import Task


@KeyDirective(fields="id")
class UserExtension(FederationType, schema_name="User"):
    id = FederationField(int)

    # Populated by the Users subgraph.
    timezone = FederationField(str) @ ExternalDirective()

    # "Overdue" depends on the user's local time, so we need `timezone` resolved first.
    overdue_task_count = FederationField(int) @ RequiresDirective(fields="timezone")

    @overdue_task_count.resolve
    def resolve_overdue_task_count(self, info: GQLInfo) -> int:
        return Task.objects.filter(assigned_to_id=self.id, due_by__lt=self.timezone).count()

1
2
3
4
5
type User @key(fields: "id") {
  id: Int!
  timezone: String! @external
  overdueTaskCount: Int! @requires(fields: "timezone")
}

Both timezone (an @external attribute) and id (a key field) are populated by the router; only overdue_task_count needs an explicit resolver.

Stub references🔗

A stub declares an entity type this subgraph references but does not resolve — the router routes resolution requests elsewhere. Mark the key as resolvable=False, declare the key field(s) explicitly, and leave the rest of the type empty. You can then use the stub as a Field reference on a resolvable QueryType, and return a representation dict from the resolver so the router can fetch the rest of the entity from the owning subgraph:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
from undine import Field, GQLInfo, QueryType
from undine.federation import FederationField, FederationType, KeyDirective

from .models import Task


@KeyDirective(fields="id", resolvable=False)
class UserStub(FederationType, schema_name="User"):
    id = FederationField(int)


@KeyDirective(fields="id")
class TaskType(QueryType[Task]):
    id = Field()
    name = Field()
    assigned_to = Field(UserStub, nullable=True)

    @assigned_to.resolve
    def resolve_assigned_to(root: Task, info: GQLInfo) -> dict | None:
        if root.assigned_to_id is None:
            return None
        return {"id": root.assigned_to_id}

1
2
3
4
5
6
7
8
9
type TaskType @key(fields: "id") {
  id: Int!
  name: String!
  assignedTo: User
}

type User @key(fields: "id", resolvable: false) {
  id: Int!
}

The stub still appears in _service.sdl (so the router can compose it), but it does not appear in the _Entity union and no _entities dispatch is wired for it.

Permissions🔗

When the router fetches entities through the injected _entities endpoint, it bypasses your top-level query entrypoints. To keep authorization consistent, Undine invokes the entity's __permissions__ classmethod on every entity resolved through _entities, whether the entity is backed by a QueryType or by a FederationType.

For a QueryType, the same __permissions__ that runs on regular entrypoints runs here too — no extra wiring is needed.

For a FederationType, define __permissions__ on the class; the _entities resolver invokes it right after __resolve_reference__ builds the instance:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from undine import GQLInfo
from undine.exceptions import GraphQLPermissionError
from undine.federation import FederationField, FederationType, KeyDirective


@KeyDirective(fields="id")
class UserExtension(FederationType, schema_name="User"):
    id = FederationField(int)
    assigned_task_count = FederationField(int)

    @classmethod
    def __permissions__(cls, instance: "UserExtension", info: GQLInfo) -> None:
        if not info.context.user.is_authenticated:
            msg = "Only authenticated users can resolve User extensions."
            raise GraphQLPermissionError(msg)

    @assigned_task_count.resolve
    def resolve_assigned_task_count(self, info: GQLInfo) -> int:
        return 0

To gate an individual FederationField, decorate a method with @<field>.permissions. Sibling fields keep resolving independently:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from undine import GQLInfo
from undine.exceptions import GraphQLPermissionError
from undine.federation import FederationField, FederationType, KeyDirective


@KeyDirective(fields="id")
class UserExtension(FederationType, schema_name="User"):
    id = FederationField(int)
    assigned_task_count = FederationField(int)

    @assigned_task_count.resolve
    def resolve_assigned_task_count(self, info: GQLInfo) -> int:
        return 0

    @assigned_task_count.permissions
    def assigned_task_count_permissions(self, info: GQLInfo, value: int) -> None:
        if not info.context.user.is_staff:
            msg = "Only staff can see the assigned task count."
            raise GraphQLPermissionError(msg)

Both hooks may also be defined async. If a permission check fails, the failing entity slot or field becomes null with the error attached, leaving the rest of the batch intact.

Apollo compatibility🔗

Undine is verified against Apollo's subgraph compatibility harness, which composes a subgraph with Apollo's users and inventory subgraphs plus a router and runs Apollo's canonical query set against the composed supergraph.

A reference implementation of Apollo's products subgraph in Undine lives at tests/test_federation/compatibility/. It contains a small Django project you can read end-to-end to see entities, FederationType extensions, @key/@requires/@provides/@override, custom directives via @composeDirective, and reference resolvers wired against real models. See its README.md for how to boot it locally and run Apollo's compliance runner.

Federation 1 Support Federation 2 Support
_service🟢
@key (single)🟢
@key (multi)🟢
@key (composite)🟢
repeatable @key🟢
@requires🟢
@provides🟢
federated tracing🟢
@link🟢
@shareable🟢
@tag🟢
@override🟢
@inaccessible🟢
@composeDirective🟢
@interfaceObject🟢

Federated tracing🔗

Undine ships FederatedTracingHook, a lifecycle hook that implements Apollo's Federated Tracing v1 (ftv1) protocol. Register it in ADDITIONAL_LIFECYCLE_HOOKS to opt in:

1
2
3
4
5
UNDINE = {
    "ADDITIONAL_LIFECYCLE_HOOKS": [
        "undine.federation.tracing.FederatedTracingHook",
    ],
}

The hook requires the protobuf package, which is not installed by default. Install it via the federation-tracing extra:

1
pip install 'undine[federation-tracing]'

When a request carries the apollo-federation-include-trace: ftv1 header, the hook records timings for every resolver call and attaches a message under extensions.ftv1 on the response. See Apollo docs for more information.