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