Lifecycle Hooks🔗

In this section, we'll cover Undine's lifecycle hooks, which allow you to hook into the execution of a GraphQL request.

LifecycleHook🔗

A GraphQL operation is executed in a series of steps. These steps are:

Parsing the GraphQL source document to a GraphQL AST.
Validation of the GraphQL AST against the GraphQL schema.
Execution of the GraphQL operation according to the GraphQL AST.

LifecycleHooks allow you to hook into the these steps. To implement a hook, you need to create a class that inherits from LifecycleHook and implement the the appropriate methods based on the steps you want to hook into. The points you can hook into are:

on_operation / on_operation_async: Encompasses the entire GraphQL operation.

from collections.abc import AsyncGenerator, Generator

from undine.hooks import LifecycleHook


class ExampleHook(LifecycleHook):
    """Example hook"""

    def on_operation(self) -> Generator[None, None, None]:
        print("before")
        yield
        print("after")

    # Async hook uses synchronous version if not implemented.
    async def on_operation_async(self) -> AsyncGenerator[None, None]:
        print("before async")
        yield
        print("after async")

on_parse / on_parse_async: Encompasses the parsing step.

from collections.abc import AsyncGenerator, Generator

from undine.hooks import LifecycleHook


class ExampleHook(LifecycleHook):
    """Example hook"""

    def on_parse(self) -> Generator[None, None, None]:
        print("before")
        yield
        print("after")

    # Async hook uses synchronous version if not implemented.
    async def on_parse_async(self) -> AsyncGenerator[None, None]:
        print("before async")
        yield
        print("after async")

on_validation / on_validation_async: Encompasses the validation step.

from collections.abc import AsyncGenerator, Generator

from undine.hooks import LifecycleHook


class ExampleHook(LifecycleHook):
    """Example hook"""

    def on_validation(self) -> Generator[None, None, None]:
        print("before")
        yield
        print("after")

    # Async hook uses synchronous version if not implemented.
    async def on_validation_async(self) -> AsyncGenerator[None, None]:
        print("before async")
        yield
        print("after async")

on_execution / on_execution_async: Encompasses the execution step.

from collections.abc import AsyncGenerator, Generator

from undine.hooks import LifecycleHook


class ExampleHook(LifecycleHook):
    """Example hook"""

    def on_execution(self) -> Generator[None, None, None]:
        print("before")
        yield
        print("after")

    # Async hook uses synchronous version if not implemented.
    async def on_execution_async(self) -> AsyncGenerator[None, None]:
        print("before async")
        yield
        print("after async")

resolve: Encompasses each field resolver (see graphql-core custom middleware).

from collections.abc import Awaitable
from typing import Any

from graphql import GraphQLFieldResolver
from graphql.pyutils import AwaitableOrValue

from undine import GQLInfo
from undine.hooks import LifecycleHook


class ExampleHook(LifecycleHook):
    """Example hook"""

    # The 'resolve' step only has a synchronous interface.
    # If you want your hook to also support awaitable resolvers,
    # you need to check if the resolver returns an awaitable and handle it separately.

    def resolve(self, resolver: GraphQLFieldResolver, root: Any, info: GQLInfo, **kwargs: Any) -> AwaitableOrValue[Any]:
        print("before")
        result = resolver(root, info, **kwargs)
        print("after")

        if info.is_awaitable(result):
            return self.resolve_async(resolver=result, root=root, info=info, **kwargs)

        return result

    async def resolve_async(self, awaitable: Awaitable[Any], root: Any, info: GQLInfo, **kwargs: Any) -> Any:
        print("before async")
        result = await awaitable
        print("after async")
        return result

Registering hooks🔗

Created hooks need to be registered using the ADDITIONAL_LIFECYCLE_HOOKS setting.

UNDINE = {
    "ADDITIONAL_LIFECYCLE_HOOKS": [
        "myproj.hooks.TimingHook",
    ],
}

Priority🔗

When multiple hooks run logic on the same step, they run in the order set by their priority. A hook with a lower priority runs its "before" portion first and its "after" portion last, so it wraps the hooks with a higher priority. You can think of them as a stack of context managers.

A hook has a priority of 1000 unless it sets one, which places it inside all of the built-in hooks. To run a hook somewhere else, set the priority class attribute.

from collections.abc import Generator
from typing import ClassVar

from undine.hooks import HookPriority, LifecycleHook


class RequestLoggingHook(LifecycleHook):
    """Runs outside every built-in hook, so it also sees the responses served from a cache."""

    priority: ClassVar[int] = HookPriority.TRACING - 10

    def on_operation(self) -> Generator[None, None, None]:
        yield

These are the priorities of the built-in hooks, available as undine.hooks.HookPriority.

TRACING (100): The OpenTelemetry, Datadog, Sentry and federated tracing hooks.
PARSE_CACHE (200): undine.hooks.ParseCacheHook.
VALIDATION_CACHE (300): undine.hooks.ValidationCacheHook.
RESPONSE_CACHE (400): undine.hooks.RequestCacheHook.
VISIBILITY_CACHE (500): undine.hooks.VisibilityCacheHook.
ATOMIC_MUTATION (600): undine.hooks.AtomicMutationHook.
PERSISTED_QUERIES (700): undine.hooks.AutomaticPersistedQueriesHook.
DEFAULT (1000): Every other hook.

LifecycleHookContext🔗

Each hook is passed a LifecycleHookContext object (self.context), which contains information about the current state of the GraphQL request. This includes:

source: Source GraphQL document string.
document: Parsed GraphQL AST. Available after parsing is complete.
validation_errors: Errors found when validating the GraphQL document. Available after validation is complete. Adding errors to this in a LifecycleHook will skip validation and exit the operation early with those errors.
variables: Variables passed to the GraphQL operation.
operation_name: The name of the GraphQL operation to run from the document. Can be empty if there is only one operation in the document.
extensions: GraphQL operation extensions received from the client.
request: Django request during which the GraphQL operation is being executed.
result: Execution result of the GraphQL operation. Adding a result to this in a LifecycleHook will cause the operation to exit early with the result.
lifecycle_hooks: LifecycleHooks in use for this operation.

Examples🔗

Here's some more complex examples of possible lifecycle hooks.

import json
from collections.abc import Generator

from django.core.cache import cache
from graphql import ExecutionResult

from undine.hooks import LifecycleHook


class CachingHook(LifecycleHook):
    """Cache execution results."""

    TIMEOUT = 60

    def on_operation(self) -> Generator[None, None, None]:
        cache_key = f"undine:{self.context.source}:{json.dumps(self.context.variables)}:{self.context.request.user.pk}"
        was_cached = False

        # Check if the result is already cached.
        if cache_key in cache:
            data = cache.get(cache_key)
            was_cached = True

            # Setting results early will cause the hooking point to not run
            # and the graphql execution to exit early with this result.
            self.context.result = ExecutionResult(data=data)

        yield

        # If results where cached, the hooking point will not run, but the
        # hook's "after" portion will. Therefore, don't re-cache the result
        # if it was already cached.
        if was_cached:
            return

        if self.context.result is not None and self.context.result.data is not None:
            cache.set(cache_key, self.context.result.data, timeout=self.TIMEOUT)

from collections.abc import Generator
from time import perf_counter_ns
from typing import Any

from graphql import GraphQLFieldResolver

from undine import GQLInfo
from undine.hooks import LifecycleHook, LifecycleHookContext


class TimingHook(LifecycleHook):
    """Time the execution of each step of the GraphQL operation."""

    def __init__(self, context: LifecycleHookContext) -> None:
        super().__init__(context)

        self.parse_timing: float | None = None
        self.validation_timing: float | None = None
        self.execution_timing: float | None = None
        self.resolver_timings: dict[str, float] = {}

    def on_operation(self) -> Generator[None, None, None]:
        start = perf_counter_ns()
        try:
            yield
        finally:
            end = perf_counter_ns()

            timings = {
                "operation": end - start,
                "parse": self.parse_timing,
                "validation": self.validation_timing,
                "execution": self.execution_timing,
                "resolvers": self.resolver_timings,
            }

            if self.context.result is not None:
                self.context.result.extensions["timings"] = timings

    def on_parse(self) -> Generator[None, None, None]:
        start = perf_counter_ns()
        try:
            yield
        finally:
            self.parse_timing = perf_counter_ns() - start

    def on_validation(self) -> Generator[None, None, None]:
        start = perf_counter_ns()
        try:
            yield
        finally:
            self.validation_timing = perf_counter_ns() - start

    def on_execution(self) -> Generator[None, None, None]:
        start = perf_counter_ns()
        try:
            yield
        finally:
            self.execution_timing = perf_counter_ns() - start

    def resolve(self, resolver: GraphQLFieldResolver, root: Any, info: GQLInfo, **kwargs: Any) -> Any:
        start = perf_counter_ns()
        try:
            return resolver(root, info, **kwargs)
        finally:
            key = ".".join(str(key) for key in info.path.as_list())
            self.resolver_timings[key] = perf_counter_ns() - start