---
description: Documentation on lifecycle hooks in Undine.
---

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

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

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

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

```python
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]{:target="_blank"}).

[custom middleware]: https://graphql-core-3.readthedocs.io/en/latest/diffs.html#custom-middleware

```python
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`](settings.md#additional_lifecycle_hooks) setting.

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

```python
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](integrations.md#opentelemetry),
  [Datadog](integrations.md#datadog), [Sentry](integrations.md#sentry) and
  [federated tracing](federation.md#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.

```python
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)
```

```python
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
```