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:

  1. Parsing the GraphQL source document to a GraphQL AST.
  2. Validation of the GraphQL AST against the GraphQL schema.
  3. Execution of the GraphQL operation.

LifecycleHooks allow you to hook into the these steps to run custom logic before or after each of these steps, or before and after the whole operation. Here is a basic example of a LifecycleHook.

from collections.abc import Generator

from undine.hooks import LifecycleHook


class ExampleHook(LifecycleHook):
    def run(self) -> Generator[None, None, None]:
        print("before")
        yield
        print("after")

To implement a hook, you need to create a class that inherits from LifecycleHook and implement the run method. run should be a generator function that yields once, marking the point in which the hook should run. Anything before the yield statement will be executed before the hooking point, and anything after the yield statement will be executed after the hooking point.

You can add this hook the different steps using settings:

1
2
3
4
5
6
UNDINE = {
    "PARSE_HOOKS": ["path.to.ExampleHook"],
    "VALIDATION_HOOKS": ["path.to.ExampleHook"],
    "EXECUTION_HOOKS": ["path.to.ExampleHook"],
    "OPERATION_HOOKS": ["path.to.ExampleHook"],
}

When multiple hooks are registered for the same hooking point, they will be run in the order they are registered. This means that 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 document AST. Available after parsing is complete.
  • variables: Variables passed to the GraphQL operation.
  • operation_name: The name of the GraphQL operation.
  • extensions: GraphQL operation extensions received from the client.
  • request: Django request during which the GraphQL request is being executed.
  • result: Execution result of the GraphQL operation.

Examples🔗

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

from collections.abc import Generator

from graphql import ExecutionResult, GraphQLError

from undine.hooks import LifecycleHook


class ErrorEnrichmentHook(LifecycleHook):
    """Catch errors and enrich them with status codes and error codes."""

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

        # Catch all exceptions raised in the hooking point and turn them into ExecutionResults.
        except Exception as err:
            msg = str(err)
            extensions = {"status_code": 500, "error_code": "INTERNAL_SERVER_ERROR"}
            new_error = GraphQLError(msg, extensions=extensions)
            self.context.result = ExecutionResult(errors=[new_error])
            return

        if self.context.result is None or self.context.result.errors is None:
            return

        # Enrich errors with status codes and error codes.
        for error in self.context.result.errors:
            if error.extensions is None:
                error.extensions = {}

            error.extensions.setdefault("status_code", 400)
            error.extensions.setdefault("error_code", "INTERNAL_SERVER_ERROR")
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 run(self) -> Generator[None, None, None]:
        cache_key = f"{self.context.source}:{json.dumps(self.context.variables)}"
        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)