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 typing import Any

from graphql import GraphQLFieldResolver

from undine import GQLInfo
from undine.hooks import LifecycleHook


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

    # The 'resolve' hook only has a synchronous interface.
    def resolve(self, resolver: GraphQLFieldResolver, root: Any, info: GQLInfo, **kwargs: Any) -> Any:
        print("before")
        result = resolver(root, info, **kwargs)
        print("after")
        return result

Created hooks need to be registered using the LIFECYCLE_HOOKS setting. When there are multiple hooks that run logic on the same step, they will be run in the order they are added in the LIFECYCLE_HOOKS setting list. Specifically, the first hook registered will have its "before" portion run first and its "after" portion run last. You can think of them as a stack of context managers.

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.
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