Federation๐Ÿ”—

Undine provides opt-in 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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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")],
)
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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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:

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:

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:

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:

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.

Visibility๐Ÿ”—

This is an experimental feature that needs to be enabled using the EXPERIMENTAL_VISIBILITY_CHECKS setting.

You can hide a FederationType from certain users by using the __is_visible__ method. Selecting a hidden FederationType results in an error that looks exactly like the type didn't exist in the first place.

from undine.federation import FederationField, FederationType, KeyDirective
from undine.typing import DjangoRequestProtocol


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

    @classmethod
    def __is_visible__(cls, request: DjangoRequestProtocol) -> bool:
        return request.user.is_superuser

You can hide a FederationField from certain users by decorating a method with the <field_name>.visible decorator. Hiding a FederationField means that trying to select it results in an error that looks exactly like the field didn't exist in the first place.

from undine.federation import FederationField, FederationType, KeyDirective
from undine.typing import DjangoRequestProtocol


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

    @assigned_task_count.visible
    def assigned_task_count_visible(self: FederationField, request: DjangoRequestProtocol) -> bool:
        return request.user.is_staff

Note that visibility expects clients to introspect and query the your schema directly. In a federation scenario, the composed supergraph that the router advertises does not change per request. This results in hidden types and fields showing when they shouldn't. This also applies to other schema entities where visibility can be used, see visibility rule.

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 LIFECYCLE_HOOKS to opt in:

1
2
3
4
5
6
7
UNDINE = {
    "LIFECYCLE_HOOKS": [
        "undine.hooks.RequestCacheHook",
        "undine.hooks.AtomicMutationHook",
        "undine.federation.tracing.FederatedTracingHook",
    ],
}

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

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.