Integrations🔗

In this section, we'll cover the integrations to other libraries that Undine includes.

Channels🔗

1	`pip install undine[channels]`

Undine provides support for GraphQL over WebSocket and GraphQL over SSE (Single Connection mode) by integrating with the channels library. Using the channels integration requires turning on Undine's Async Support.

You'll need to configure Django in your project's asgi.py file so that requests are sent to Undine's channels consumers. There are three app wrappers available depending on which protocols you need.

For WebSocket support only:

import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "...")
django_application = get_asgi_application()

# Needs be imported after 'django_application' is created!
from undine.integrations.channels import get_websocket_enabled_app

application = get_websocket_enabled_app(django_application)

For SSE Single Connection mode only:

import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "...")
django_application = get_asgi_application()

# Needs be imported after 'django_application' is created!
from undine.integrations.channels import get_sse_enabled_app

application = get_sse_enabled_app(django_application)

For both WebSocket and SSE Single Connection mode:

import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "...")
django_application = get_asgi_application()

# Needs be imported after 'django_application' is created!
from undine.integrations.channels import get_websocket_and_sse_enabled_app

application = get_websocket_and_sse_enabled_app(django_application)

The integration also provides ChannelLayerSubscriptionBroker, which delivers signal subscription events between worker processes over a channel layer.

GraphiQL🔗

Undine includes a built-in GraphiQL interface for exploring and testing your GraphQL API. You can enable it using the GRAPHIQL_ENABLED setting. You should also set ALLOW_INTROSPECTION_QUERIES to True so that GraphiQL can introspect the schema. GraphiQL is then accessible by navigating to the GraphQL endpoint in a browser.

GraphiQL includes the explorer and history plugins. The URL encodes the current document, variables, and headers, so it can be shared with others.

By default, subscriptions use WebSockets. To use Server-Sent Events instead, use the GRAPHIQL_SSE_ENABLED setting. Single connection mode can be enabled with the GRAPHIQL_SSE_SINGLE_CONNECTION setting.

1	`pip install undine[debug]`

Undine integrates with django-debug-toolbar by modifying the toolbar HTML so that it integrates with GraphiQL. After installing django-debug-toolbar, Undine should automatically patch it without any additional configuration.

Django Model Translation🔗

Undine integrates with django-modeltranslation by allowing you to modify how autogenerated Fields, Inputs, Filters and Orders are created. Specifically, this happens using two settings: MODELTRANSLATION_INCLUDE_TRANSLATABLE and MODELTRANSLATION_INCLUDE_TRANSLATIONS.

Let's say you the following Model (with MODELTRANSLATION_LANGUAGES = ("en", "fi"))

from django.db import models


class Task(models.Model):
    name = models.CharField(max_length=255)

    # Created by modeltranslation
    name_en: str | None
    name_fi: str | None

...and the following translation options

from modeltranslation.decorators import register
from modeltranslation.translator import TranslationOptions

from .models import Task


@register(Task)
class TaskTranslationOptions(TranslationOptions):
    fields = ["name"]

Based on the Model's translation options, django-modeltranslation adds additional fields for each language defined by the MODELTRANSLATION_LANGUAGES setting. Let's call the added fields the "translatable" fields, and the fields they are based on the "translation" fields.

Using the MODELTRANSLATION_INCLUDE_TRANSLATABLE and MODELTRANSLATION_INCLUDE_TRANSLATIONS settings, you can control which of these fields undine will add to your schema when using autogeneration. By default, only the translation fields are added. You can of course always add the translatable fields manually.

Note that due to the way that django-modeltranslation works, the translation fields are always nullable, even for the default language.

OpenTelemetry🔗

1	`pip install undine[opentelemetry]`

Undine ships OpenTelemetryHook, a lifecycle hook that records an OpenTelemetry span for each GraphQL operation, with a child span for the parsing, validation and execution steps. Register it in ADDITIONAL_LIFECYCLE_HOOKS to opt in:

UNDINE = {
    "ADDITIONAL_LIFECYCLE_HOOKS": [
        "undine.integrations.opentelemetry.OpenTelemetryHook",
    ],
}

The operation span is named after the operation, e.g. query FindTask, and carries the OpenTelemetry semantic conventions for GraphQL: graphql.operation.name, graphql.operation.type and graphql.document. If the operation results in errors, the span status is set to ERROR and each error is recorded on the span.

For a subscription, the operation span lasts as long as the subscription does. Each result sent to the client gets its own execution span inside that same trace.

Undine depends on opentelemetry-api only. Your application configures the SDK, which is the standard split for instrumented libraries.

Field spans🔗

OpenTelemetryFullHook records everything OpenTelemetryHook does, plus a span for each resolved field. This is opt-in, since a span per field is expensive on large responses. Register it instead of OpenTelemetryHook when you need per-field timings:

UNDINE = {
    "ADDITIONAL_LIFECYCLE_HOOKS": [
        "undine.integrations.opentelemetry.OpenTelemetryFullHook",
    ],
}

Introspection queries are recorded without their field spans. An introspection query resolves a field for every type and field in the schema, so a span for each one buries the operations that say something about the service.

To decide yourself, set the OPENTELEMETRY_SKIP_FIELD_SPANS_PREDICATE setting to a function that takes the lifecycle hook context and returns whether the field spans should be left out. Undine ships undine.integrations.opentelemetry.never_skip_field_spans for recording every field span.

Sensitive data🔗

A GraphQL request can carry sensitive data in two places, the document and the variables. The values from both are kept out of the spans by default.

Arguments a client hardcodes in the document are redacted before the document is recorded. The structure of the document is kept, so traces can still be grouped by operation shape:

query FindUser {
  user(email: "***") {
    name
  }
}

Variables are recorded by name, with each value replaced with ***:

{"email": "***", "first": "***"}

Only the top-level keys are kept, since the keys inside a variable can be client data as well, e.g. when the variable is typed as a JSON scalar. To record the values of some variables, set the OPENTELEMETRY_VARIABLES_CALLBACK setting to a function that returns the variables you want to record.

from typing import Any

from undine.hooks import LifecycleHookContext

SAFE_VARIABLES = {"first", "last", "offset"}

def traced_variables(context: LifecycleHookContext) -> dict[str, Any]:
    return {key: value for key, value in context.variables.items() if key in SAFE_VARIABLES}

Undine ships undine.integrations.opentelemetry.no_traced_variables for recording no variables at all.

Custom attributes🔗

To add your own attributes to the spans, set the OPENTELEMETRY_SPAN_CALLBACK setting to a function. It's called for every span the hook records: the operation span, the parse, validation and execution spans, and each field span. For the operation span it runs once the span is fully described (name, type and document set) and the operation has finished executing, so it can also react to the result, e.g. to record a custom attribute when the operation failed:

from opentelemetry.trace import Span

from undine.hooks import LifecycleHookContext


def tag_with_error_count(span: Span, context: LifecycleHookContext) -> None:
    error_count = len(context.result.errors or []) if context.result is not None else 0  # type: ignore[union-attr]
    span.set_attribute("graphql.error_count", error_count)

Sentry and Datadog🔗

Both vendors accept OpenTelemetry data, but with caveats.

Sentry's OTLP endpoint is in open beta and drops span events. It also gives you traces only, while the dedicated Sentry integration below turns failing operations into issues. Prefer that one.

Datadog users on the OpenTelemetry SDK lose Continuous Profiler, App & API Protection, Data Streams Monitoring, RUM correlation and Source Code Integration. Datadog users who prefer the OpenTelemetry path can set DD_TRACE_OTEL_ENABLED=true, which makes ddtrace serve the OpenTelemetry API this hook is written against. Prefer the dedicated Datadog integration below unless you have a reason to use OpenTelemetry directly.

Datadog🔗

1	`pip install undine[datadog]`

Undine ships DatadogHook, a lifecycle hook that records a native Datadog span for each GraphQL operation, with a child span for the parsing, validation and execution steps. Prefer this over the OpenTelemetry hook for Datadog: instrumenting natively keeps Continuous Profiler, App & API Protection, Data Streams Monitoring, RUM correlation and Source Code Integration working, none of which survive the OpenTelemetry SDK path. Register it in ADDITIONAL_LIFECYCLE_HOOKS to opt in:

UNDINE = {
    "ADDITIONAL_LIFECYCLE_HOOKS": [
        "undine.integrations.datadog.DatadogHook",
    ],
}

Every span has span_type set to "graphql". The operation span is named after the operation, e.g. query FindTask, and its resource is <operation name>:<query hash> (or just the hash for an anonymous operation) — this is Datadog's primary grouping dimension, so getting it right is what keeps traces for the same operation grouped together. The operation span also carries the graphql.operation.name and graphql.operation.type tags.

The service name defaults to "undine". Set the DATADOG_SERVICE_NAME setting to change it, e.g. when several services report to the same Datadog account.

Field spans🔗

DatadogFullHook records everything DatadogHook does, plus a span for each resolved field, tagged with graphql.field.name, graphql.field.parent.type, graphql.field.path and graphql.path. This is opt-in, since a span per field is expensive on large responses. Register it instead of DatadogHook when you need per-field timings:

UNDINE = {
    "ADDITIONAL_LIFECYCLE_HOOKS": [
        "undine.integrations.datadog.DatadogFullHook",
    ],
}

Introspection queries are recorded without their field spans. An introspection query resolves a field for every type and field in the schema, so a span for each one buries the operations that say something about the service.

To decide yourself, set the DATADOG_SKIP_FIELD_SPANS_PREDICATE setting to a function that takes the lifecycle hook context and returns whether the field spans should be left out. Undine ships undine.integrations.datadog.never_skip_field_spans for recording every field span.

Sensitive data🔗

A GraphQL request can carry sensitive data in two places, the document and the variables. The values from both are kept out of the spans by default.

Arguments a client hardcodes in the document are redacted before the document is recorded. The structure of the document is kept, so traces can still be grouped by operation shape:

query FindUser {
  user(email: "***") {
    name
  }
}

Variables are recorded by name, with each value replaced with ***:

{"email": "***", "first": "***"}

Only the top-level keys are kept, since the keys inside a variable can be client data as well, e.g. when the variable is typed as a JSON scalar. To record the values of some variables, set the DATADOG_VARIABLES_CALLBACK setting to a function that returns the variables you want to record.

from typing import Any

from undine.hooks import LifecycleHookContext

SAFE_VARIABLES = {"first", "last", "offset"}

def traced_variables(context: LifecycleHookContext) -> dict[str, Any]:
    return {key: value for key, value in context.variables.items() if key in SAFE_VARIABLES}

Undine ships undine.integrations.datadog.no_traced_variables for recording no variables at all.

Custom attributes🔗

To add your own tags to the spans, set the DATADOG_SPAN_CALLBACK setting to a function. It's called for every span the hook records: the operation span, the parse, validation and execution spans, and each field span. For the operation span it runs once the span is fully described (type, name and resource set) and the operation has finished executing, so it can also react to the result, e.g. to add a tag when the operation failed:

from ddtrace.trace import Span

from undine.hooks import LifecycleHookContext


def tag_with_error_count(span: Span, context: LifecycleHookContext) -> None:
    error_count = len(context.result.errors or []) if context.result is not None else 0  # type: ignore[union-attr]
    span.set_tag("graphql.error_count", error_count)

Sentry🔗

1	`pip install undine[sentry]`

Undine ships SentryHook, a lifecycle hook that instruments GraphQL operations for Sentry. Prefer this over sending OpenTelemetry data to Sentry's OTLP endpoint, which is in open beta, drops span events, and gives you traces only. Register it in ADDITIONAL_LIFECYCLE_HOOKS to opt in:

UNDINE = {
    "ADDITIONAL_LIFECYCLE_HOOKS": [
        "undine.integrations.sentry.SentryHook",
    ],
}

The hook does four things.

It starts a transaction when nothing else has. Sentry only records spans that belong to a transaction. Its Django integration starts one for an HTTP request, but not for the WebSocket and SSE connections that subscriptions run on, so the hook starts one for those itself. Without it, nothing about a subscription reaches Sentry.

It names the transaction after the GraphQL operation. Sentry's Django integration names the transaction after the HTTP route, so without this every GraphQL request in your service collapses into a single /graphql/ transaction. The hook renames it to the operation name, e.g. FindTask, and sets the transaction operation to graphql.query, graphql.mutation or graphql.subscription. Anonymous operations keep the route name, since they have no name to use.

It records spans. One span for the operation, e.g. query FindTask, with a child span for the parsing, validation and execution steps. The operation span carries the graphql.operation.name, graphql.operation.type and graphql.document data.

It reports failing operations as issues. Each reported error becomes a Sentry issue with the GraphQL context attached, so an exception raised in a resolver is reported with the exception that actually failed, not with the generic message the client receives from error masking.

Undine's own log records don't become Sentry issues while this hook is installed. Undine logs the errors that it masks, and the hook already reports those failures with more detail, so letting Sentry's logging integration report them again would only create a second, poorer issue for the same failure.

Which errors are reported🔗

By default, only errors that indicate a fault in the server are reported. A client mistake, like a validation error or a denied permission, is not an incident, so reporting those would only add noise.

To decide yourself, set the SENTRY_REPORT_ERROR_PREDICATE setting to a function. Undine ships undine.integrations.sentry.report_all_errors for reporting every GraphQL error.

from http import HTTPStatus

from graphql import GraphQLError

IGNORED_ERROR_CODES = {"PERMISSION_DENIED", "VALIDATION_ERROR"}


def should_report_error(error: GraphQLError) -> bool:
    if error.extensions.get("error_code") in IGNORED_ERROR_CODES:
        return False
    return error.extensions.get("status_code") == HTTPStatus.INTERNAL_SERVER_ERROR

Field spans🔗

SentryFullHook records everything SentryHook does, plus a span for each resolved field, with the graphql.field.name, graphql.field.parent.type, graphql.field.path and graphql.path data. This is opt-in, since a span per field is expensive on large responses. Register it instead of SentryHook when you need per-field timings:

UNDINE = {
    "ADDITIONAL_LIFECYCLE_HOOKS": [
        "undine.integrations.sentry.SentryFullHook",
    ],
}

Introspection queries are recorded without their field spans. An introspection query resolves a field for every type and field in the schema, so a span for each one buries the operations that say something about the service.

To decide yourself, set the SENTRY_SKIP_FIELD_SPANS_PREDICATE setting to a function that takes the lifecycle hook context and returns whether the field spans should be left out. Undine ships undine.integrations.sentry.never_skip_field_spans for recording every field span.

Span streaming🔗

Sentry has two span APIs, and a client can only use the one its trace_lifecycle option selects. The hook records through whichever one is in use, so everything above holds in either mode. In span streaming mode each span is sent on its own, and the name of the trace is carried by the segment instead of a transaction. Sentry's span streaming API is experimental, so treat this the same way you treat the option itself.

Sensitive data🔗

The graphql.document span data is always redacted, so a value a client hardcodes in the document never reaches a span. The structure of the document is kept, so traces can still be grouped by operation shape:

query FindUser {
  user(email: "***") {
    name
  }
}

The graphql.variables span data carries the name of each variable with its value replaced with ***. Only the top-level keys are kept, since the keys inside a variable can be client data as well, e.g. when the variable is typed as a JSON scalar. To record the values of some variables, set the SENTRY_VARIABLES_CALLBACK setting to a function that returns the variables you want to record. Undine ships undine.integrations.sentry.no_traced_variables for recording no variables at all.

An issue is a separate payload from the spans, and Sentry doesn't show span data on it, so the hook attaches the operation to the issue as well. There it carries the operation name, the redacted document and the redacted variables. Enable Sentry's own [send_default_pii]{:target="_blank"} option to attach the document as the client wrote it, with the values of the variables. This is the SDK's own control, so it covers Undine's data the same way it covers the rest of your application.

Custom attributes🔗

To add your own attributes to the spans, set the SENTRY_SPAN_CALLBACK setting to a function. It's called for every span the hook records: the operation span, the parse, validation and execution spans, and each field span. For the operation span it runs once the span is fully described (name, type and document set) and the operation has finished executing, so it can also react to the result, e.g. to record an attribute when the operation failed:

from undine.hooks import LifecycleHookContext
from undine.integrations.sentry import RecordedSpan


def tag_with_error_count(span: RecordedSpan, context: LifecycleHookContext) -> None:
    error_count = len(context.result.errors or []) if context.result is not None else 0  # type: ignore[union-attr]
    span.set_data("graphql.error_count", error_count)

The callback receives a RecordedSpan, which is Undine's view over the two span APIs, so set_data works in either mode. Reach through span.sentry_span for the things the two APIs don't share, such as setting a status.

Mypy🔗

Undine ships a mypy plugin that adds additional static type checking for types defined in Undine. To enable it, add the following to your mypy.ini file:

[mypy]
plugins = mypy_undine

The plugin adds the following additional type checks:

If there is a check that you think should be included, please open an issue or a pull request!

Pytest🔗

Undine comes with a pytest plugin that includes a testing client and few fixtures to help you write tests for your GraphQL APIs.

The GraphQLClient class is wrapper around Django's test client that makes testing your GraphQL API easier. It can be added to a test using the graphql fixture. Here is a simple example:

def test_example(graphql) -> None:
    query = "query { test }"

    response = graphql(query)

    assert response.data == {"hello": "Hello, World!"}

GraphQL requests can be made by calling the client as shown above. This makes a request to the GraphQL endpoint set by the GRAPHQL_PATH setting.

GraphQL variables can be passed using the variables argument. If these variables include any files, the client will automatically create a GraphQL multipart request instead of a normal GraphQL request.

def test_example(graphql) -> None:
    mutation = "mutation($input: TestInput!) { test(input: $input) }"
    data = {"name": "World"}

    response = graphql(mutation, variables={"input": data})

    assert response.data == {"hello": "Hello, World!"}

The client returns a custom response object GraphQLClientResponse, which has a number of useful properties for introspecting the response. The response object also has details on the database queries that were executed during the request, which can be useful for debugging the performance of your GraphQL API.

def test_example(graphql) -> None:
    query = "query { test { edges { node { id } } } }"

    response = graphql(query, count_queries=True)

    # The whole response
    assert response.json == {
        "data": {"test": {"edges": [{"node": {"id": "1"}}]}},
        "errors": [{"message": "Error message", "path": ["test"]}],
    }

    # Error properties
    assert response.has_errors is True
    assert response.errors == [{"message": "Error message", "path": ["test"]}]
    assert response.error_message(0) == "Error message"

    # Data properties
    assert response.data == {"test": {"edges": [{"node": {"id": "1"}}]}}
    assert response.results == {"edges": [{"node": {"id": "1"}}]}

    # Connection specific properties
    assert response.edges == [{"node": {"id": "1"}}]
    assert response.node(0) == {"id": "1"}

    # Check queries (requires `count_queries=True`)
    assert response.query_count == 1
    assert response.queries == ["SELECT 1;"]
    response.assert_query_count(1)

An async version of the client is also available, which can be accessed from the graphql_async fixture.

import pytest


@pytest.mark.asyncio  # Requires the `pytest-asyncio` plugin
@pytest.mark.django_db(transaction=True)  # For sessions
async def test_example(graphql_async) -> None:
    query = "query { test }"

    response = await graphql_async(query)

    assert response.data == {"hello": "Hello, World!"}

The plugin also includes a undine_settings fixture that allows modifying Undine's settings during testing more easily.

def test_example(undine_settings) -> None:
    undine_settings.NO_ERROR_LOCATION = True

If the channels integration is installed, the test client can also send GraphQL over WebSocket requests using the over_websocket method.

import pytest


@pytest.mark.asyncio  # Requires the `pytest-asyncio` plugin
@pytest.mark.django_db(transaction=True)  # For sessions
async def test_graphql(graphql) -> None:
    query = "query { test }"
    async for response in graphql.over_websocket(query):
        assert response.data == {"test": "Hello, World!"}

Integrations🔗

Channels🔗

GraphiQL🔗

Django Debug Toolbar🔗

Django Model Translation🔗

OpenTelemetry🔗

Field spans🔗

Sensitive data🔗

Custom attributes🔗

Sentry and Datadog🔗

Datadog🔗

Field spans🔗

Sensitive data🔗

Custom attributes🔗

Sentry🔗

Which errors are reported🔗

Field spans🔗

Span streaming🔗

Sensitive data🔗

Custom attributes🔗

Mypy🔗

Pytest🔗