Skip to content

Package

nautobot_graphql_observability

App declaration for nautobot_graphql_observability.

NautobotAppGraphqlObservabilityConfig

Bases: NautobotAppConfig

App configuration for the nautobot_graphql_observability app.

Source code in nautobot_graphql_observability/__init__.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class NautobotAppGraphqlObservabilityConfig(NautobotAppConfig):
    """App configuration for the nautobot_graphql_observability app."""

    name = "nautobot_graphql_observability"
    verbose_name = "Nautobot App GraphQL Observability"
    version = __version__
    author = "Lydien SANDANASAMY"
    description = "Nautobot App GraphQL Observability."
    base_url = "nautobot-graphql-observability"
    required_settings = []
    default_settings = {
        "graphql_metrics_enabled": True,
        "track_query_depth": True,
        "track_query_complexity": True,
        "track_field_resolution": False,
        "track_per_user": True,
        "query_logging_enabled": False,
        "log_query_body": False,
        "log_query_variables": False,
    }
    middleware = [
        "nautobot_graphql_observability.django_middleware.GraphQLObservabilityDjangoMiddleware",
    ]
    docs_view_name = "plugins:nautobot_graphql_observability:docs"
    searchable_models = []

    def ready(self):
        """Patch Nautobot's GraphQLDRFAPIView to load Graphene middleware from settings.

        Nautobot's ``GraphQLDRFAPIView.init_graphql()`` does not load middleware
        from ``GRAPHENE["MIDDLEWARE"]`` when ``self.middleware`` is ``None`` (the
        default).  This is a known limitation of the DRF-based GraphQL view —
        the standard ``graphene_django.views.GraphQLView`` (used by the GraphiQL
        UI at ``/graphql/``) loads middleware correctly.

        No official extension point (``override_views``, etc.) can replace this
        patch because the ``graphql-api`` URL is registered without a namespace.
        Request duration and query logging are handled by
        :class:`~nautobot_graphql_observability.django_middleware.GraphQLObservabilityDjangoMiddleware`,
        which is registered via :attr:`middleware` (the official Nautobot mechanism).
        """
        super().ready()
        self._patch_init_graphql()

    @staticmethod
    def _patch_init_graphql():
        """Patch ``GraphQLDRFAPIView.init_graphql`` to load ``GRAPHENE["MIDDLEWARE"]``."""
        from nautobot.core.api.views import GraphQLDRFAPIView  # pylint: disable=import-outside-toplevel

        original_init_graphql = GraphQLDRFAPIView.init_graphql

        def patched_init_graphql(view_self):
            original_init_graphql(view_self)
            if view_self.middleware is None:
                from graphene_django.settings import graphene_settings  # pylint: disable=import-outside-toplevel
                from graphene_django.views import instantiate_middleware  # pylint: disable=import-outside-toplevel

                if graphene_settings.MIDDLEWARE:
                    view_self.middleware = list(instantiate_middleware(graphene_settings.MIDDLEWARE))

        GraphQLDRFAPIView.init_graphql = patched_init_graphql

ready()

Patch Nautobot's GraphQLDRFAPIView to load Graphene middleware from settings.

Nautobot's GraphQLDRFAPIView.init_graphql() does not load middleware from GRAPHENE["MIDDLEWARE"] when self.middleware is None (the default). This is a known limitation of the DRF-based GraphQL view — the standard graphene_django.views.GraphQLView (used by the GraphiQL UI at /graphql/) loads middleware correctly.

No official extension point (override_views, etc.) can replace this patch because the graphql-api URL is registered without a namespace. Request duration and query logging are handled by :class:~nautobot_graphql_observability.django_middleware.GraphQLObservabilityDjangoMiddleware, which is registered via :attr:middleware (the official Nautobot mechanism).

Source code in nautobot_graphql_observability/__init__.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def ready(self):
    """Patch Nautobot's GraphQLDRFAPIView to load Graphene middleware from settings.

    Nautobot's ``GraphQLDRFAPIView.init_graphql()`` does not load middleware
    from ``GRAPHENE["MIDDLEWARE"]`` when ``self.middleware`` is ``None`` (the
    default).  This is a known limitation of the DRF-based GraphQL view —
    the standard ``graphene_django.views.GraphQLView`` (used by the GraphiQL
    UI at ``/graphql/``) loads middleware correctly.

    No official extension point (``override_views``, etc.) can replace this
    patch because the ``graphql-api`` URL is registered without a namespace.
    Request duration and query logging are handled by
    :class:`~nautobot_graphql_observability.django_middleware.GraphQLObservabilityDjangoMiddleware`,
    which is registered via :attr:`middleware` (the official Nautobot mechanism).
    """
    super().ready()
    self._patch_init_graphql()

nautobot_graphql_observability.middleware

Graphene middleware for exporting Prometheus metrics from GraphQL queries.

PrometheusMiddleware

Graphene middleware that instruments GraphQL resolvers with Prometheus metrics.

On root-level resolutions, records counters and advanced metrics immediately (these are not timing-sensitive) and stashes operation labels onto the request so that :class:PrometheusDjangoMiddleware can record the duration histogram after the full HTTP response is built.

Optionally records advanced metrics based on app configuration:

  • track_query_depth: Record query nesting depth histogram.
  • track_query_complexity: Record query field count histogram.
  • track_field_resolution: Record per-field resolver duration histogram.
  • track_per_user: Record per-user request counter.

Usage in Django settings::

GRAPHENE = {
    "MIDDLEWARE": [
        "nautobot_graphql_observability.middleware.PrometheusMiddleware",
    ]
}
Source code in nautobot_graphql_observability/middleware.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class PrometheusMiddleware:  # pylint: disable=too-few-public-methods
    """Graphene middleware that instruments GraphQL resolvers with Prometheus metrics.

    On root-level resolutions, records counters and advanced metrics immediately
    (these are not timing-sensitive) and stashes operation labels onto the
    request so that :class:`PrometheusDjangoMiddleware` can record the duration
    histogram after the full HTTP response is built.

    Optionally records advanced metrics based on app configuration:

    - ``track_query_depth``: Record query nesting depth histogram.
    - ``track_query_complexity``: Record query field count histogram.
    - ``track_field_resolution``: Record per-field resolver duration histogram.
    - ``track_per_user``: Record per-user request counter.

    Usage in Django settings::

        GRAPHENE = {
            "MIDDLEWARE": [
                "nautobot_graphql_observability.middleware.PrometheusMiddleware",
            ]
        }
    """

    def resolve(self, next: callable, root: object, info: GraphQLResolveInfo, **kwargs: object) -> object:  # pylint: disable=redefined-builtin
        """Intercept each field resolution and record metrics.

        Root-level resolutions (root is None) record counters and advanced
        metrics and stash labels for the Django middleware to record duration.
        Nested resolutions optionally record per-field duration when enabled.

        Args:
            next (callable): Callable to continue the resolution chain.
            root (object): Parent resolved value. None for top-level fields.
            info (GraphQLResolveInfo): GraphQL resolve info containing operation metadata.
            **kwargs (object): Field arguments.

        Returns:
            object: The result of the resolver.
        """
        config = _get_app_settings()

        if root is not None:
            if config.get("track_field_resolution", False):
                return self._resolve_field_with_metrics(next, root, info, **kwargs)
            return next(root, info, **kwargs)

        operation_type = info.operation.operation.value
        operation_name = self._get_operation_name(info)

        # Stash labels on the request (only for the first root field) so
        # the Django middleware can record the full-request duration.
        # For DRF views, info.context is a DRF Request wrapping a WSGIRequest.
        # The Django middleware sees the WSGIRequest, so stash on both.
        request = info.context
        if not hasattr(request, _REQUEST_ATTR):
            meta = {
                "operation_type": operation_type,
                "operation_name": operation_name,
            }
            stash_meta_on_request(request, _REQUEST_ATTR, meta)

        try:
            result = next(root, info, **kwargs)
            return result
        except Exception as error:
            graphql_errors_total.labels(
                operation_type=operation_type,
                operation_name=operation_name,
                error_type=type(error).__name__,
            ).inc()
            # Mark the error on the stashed metadata so the Django middleware
            # records the correct status.
            meta = getattr(request, _REQUEST_ATTR, None)
            if meta is not None:
                meta["error"] = True
            raise
        finally:
            # Counters and advanced metrics are not timing-sensitive, record now.
            status = "error" if getattr(request, _REQUEST_ATTR, {}).get("error") else "success"

            graphql_requests_total.labels(
                operation_type=operation_type,
                operation_name=operation_name,
                status=status,
            ).inc()

            self._record_advanced_metrics(info, operation_name, config)

    @staticmethod
    def _resolve_field_with_metrics(next, root, info, **kwargs):  # pylint: disable=redefined-builtin
        """Resolve a nested field while recording per-field duration."""
        type_name = info.parent_type.name if info.parent_type else "Unknown"
        field_name = info.field_name

        start_time = time.monotonic()
        try:
            return next(root, info, **kwargs)
        finally:
            duration = time.monotonic() - start_time
            graphql_field_resolution_duration_seconds.labels(
                type_name=type_name,
                field_name=field_name,
            ).observe(duration)

    @staticmethod
    def _record_advanced_metrics(info, operation_name, config):
        """Record query depth, complexity, and per-user metrics if enabled."""
        if config.get("track_query_depth", True):
            depth = calculate_query_depth(info.operation.selection_set, info.fragments)
            graphql_query_depth.labels(operation_name=operation_name).observe(depth)

        if config.get("track_query_complexity", True):
            complexity = calculate_query_complexity(info.operation.selection_set, info.fragments)
            graphql_query_complexity.labels(operation_name=operation_name).observe(complexity)

        if config.get("track_per_user", True):
            user = "anonymous"
            request = info.context
            if hasattr(request, "user") and hasattr(request.user, "is_authenticated"):
                if request.user.is_authenticated:
                    user = request.user.username
            graphql_requests_by_user_total.labels(
                user=user,
                operation_type=info.operation.operation.value,
                operation_name=operation_name,
            ).inc()

    @staticmethod
    def _get_operation_name(info: GraphQLResolveInfo) -> str:
        """Extract the operation name from the GraphQL query.

        Uses the explicit operation name if provided, otherwise falls back
        to the sorted, comma-joined root field names (e.g. "devices,locations").
        """
        if info.operation.name:
            return info.operation.name.value
        root_fields = []
        if info.operation.selection_set:
            for selection in info.operation.selection_set.selections:
                if isinstance(selection, FieldNode):
                    root_fields.append(selection.name.value)
        return ",".join(sorted(root_fields)) if root_fields else "anonymous"

resolve(next, root, info, **kwargs)

Intercept each field resolution and record metrics.

Root-level resolutions (root is None) record counters and advanced metrics and stash labels for the Django middleware to record duration. Nested resolutions optionally record per-field duration when enabled.

Parameters:

Name Type Description Default
next callable

Callable to continue the resolution chain.

 required
root object

Parent resolved value. None for top-level fields.

 required
info GraphQLResolveInfo

GraphQL resolve info containing operation metadata.

 required
**kwargs object

Field arguments.

 {}

Returns:

Name Type Description
object object

The result of the resolver.
Source code in nautobot_graphql_observability/middleware.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def resolve(self, next: callable, root: object, info: GraphQLResolveInfo, **kwargs: object) -> object:  # pylint: disable=redefined-builtin
    """Intercept each field resolution and record metrics.

    Root-level resolutions (root is None) record counters and advanced
    metrics and stash labels for the Django middleware to record duration.
    Nested resolutions optionally record per-field duration when enabled.

    Args:
        next (callable): Callable to continue the resolution chain.
        root (object): Parent resolved value. None for top-level fields.
        info (GraphQLResolveInfo): GraphQL resolve info containing operation metadata.
        **kwargs (object): Field arguments.

    Returns:
        object: The result of the resolver.
    """
    config = _get_app_settings()

    if root is not None:
        if config.get("track_field_resolution", False):
            return self._resolve_field_with_metrics(next, root, info, **kwargs)
        return next(root, info, **kwargs)

    operation_type = info.operation.operation.value
    operation_name = self._get_operation_name(info)

    # Stash labels on the request (only for the first root field) so
    # the Django middleware can record the full-request duration.
    # For DRF views, info.context is a DRF Request wrapping a WSGIRequest.
    # The Django middleware sees the WSGIRequest, so stash on both.
    request = info.context
    if not hasattr(request, _REQUEST_ATTR):
        meta = {
            "operation_type": operation_type,
            "operation_name": operation_name,
        }
        stash_meta_on_request(request, _REQUEST_ATTR, meta)

    try:
        result = next(root, info, **kwargs)
        return result
    except Exception as error:
        graphql_errors_total.labels(
            operation_type=operation_type,
            operation_name=operation_name,
            error_type=type(error).__name__,
        ).inc()
        # Mark the error on the stashed metadata so the Django middleware
        # records the correct status.
        meta = getattr(request, _REQUEST_ATTR, None)
        if meta is not None:
            meta["error"] = True
        raise
    finally:
        # Counters and advanced metrics are not timing-sensitive, record now.
        status = "error" if getattr(request, _REQUEST_ATTR, {}).get("error") else "success"

        graphql_requests_total.labels(
            operation_type=operation_type,
            operation_name=operation_name,
            status=status,
        ).inc()

        self._record_advanced_metrics(info, operation_name, config)

nautobot_graphql_observability.logging_middleware

Graphene middleware for logging GraphQL queries via Python's logging module.

GraphQLQueryLoggingMiddleware

Graphene middleware that captures GraphQL query metadata for logging.

On root-level resolutions, stashes operation metadata onto the request so that :class:GraphQLQueryLoggingDjangoMiddleware can emit a log entry with the real total request duration after the full response is built.

Controlled by app settings:

  • query_logging_enabled: Master switch (default: False).
  • log_query_body: Include the full query text (default: False).
  • log_query_variables: Include query variables (default: False).

Usage in Django settings::

GRAPHENE = {
    "MIDDLEWARE": [
        "nautobot_graphql_observability.logging_middleware.GraphQLQueryLoggingMiddleware",
        "nautobot_graphql_observability.middleware.PrometheusMiddleware",
    ]
}
Source code in nautobot_graphql_observability/logging_middleware.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class GraphQLQueryLoggingMiddleware:  # pylint: disable=too-few-public-methods
    """Graphene middleware that captures GraphQL query metadata for logging.

    On root-level resolutions, stashes operation metadata onto the request so
    that :class:`GraphQLQueryLoggingDjangoMiddleware` can emit a log entry
    with the **real** total request duration after the full response is built.

    Controlled by app settings:

    - ``query_logging_enabled``: Master switch (default: False).
    - ``log_query_body``: Include the full query text (default: False).
    - ``log_query_variables``: Include query variables (default: False).

    Usage in Django settings::

        GRAPHENE = {
            "MIDDLEWARE": [
                "nautobot_graphql_observability.logging_middleware.GraphQLQueryLoggingMiddleware",
                "nautobot_graphql_observability.middleware.PrometheusMiddleware",
            ]
        }
    """

    def resolve(self, next: callable, root: object, info: GraphQLResolveInfo, **kwargs: object) -> object:  # pylint: disable=redefined-builtin
        """Intercept root-level resolutions and stash metadata on the request.

        Args:
            next (callable): Callable to continue the resolution chain.
            root (object): Parent resolved value. None for top-level fields.
            info (GraphQLResolveInfo): GraphQL resolve info containing operation metadata.
            **kwargs (object): Field arguments.

        Returns:
            object: The result of the resolver.
        """
        if root is not None:
            return next(root, info, **kwargs)

        config = _get_app_settings()
        if not config.get("query_logging_enabled", False):
            return next(root, info, **kwargs)

        # Stash metadata on the request (only for the first root field).
        # For DRF views, info.context is a DRF Request wrapping a WSGIRequest.
        # The Django middleware sees the WSGIRequest, so stash on both.
        request = info.context
        if not hasattr(request, _REQUEST_ATTR):
            meta = {
                "operation_type": info.operation.operation.value,
                "operation_name": PrometheusMiddleware._get_operation_name(info),
                "user": self._get_user(info),
                "config": config,
            }

            if config.get("log_query_body", False):
                meta["query_body"] = _extract_query_body(info)

            if config.get("log_query_variables", False):
                meta["variables"] = _extract_variables(info)

            stash_meta_on_request(request, _REQUEST_ATTR, meta)

        try:
            return next(root, info, **kwargs)
        except Exception as error:
            # Record the error on the stashed metadata so the Django
            # middleware can log it.
            meta = getattr(request, _REQUEST_ATTR, None)
            if meta is not None:
                meta["error"] = error
            raise

    @staticmethod
    def _get_user(info):
        """Extract the username from the request context."""
        request = info.context
        if hasattr(request, "user") and hasattr(request.user, "is_authenticated"):
            if request.user.is_authenticated:
                return request.user.username
        return "anonymous"

resolve(next, root, info, **kwargs)

Intercept root-level resolutions and stash metadata on the request.

Parameters:

Name Type Description Default
next callable

Callable to continue the resolution chain.

 required
root object

Parent resolved value. None for top-level fields.

 required
info GraphQLResolveInfo

GraphQL resolve info containing operation metadata.

 required
**kwargs object

Field arguments.

 {}

Returns:

Name Type Description
object object

The result of the resolver.
Source code in nautobot_graphql_observability/logging_middleware.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def resolve(self, next: callable, root: object, info: GraphQLResolveInfo, **kwargs: object) -> object:  # pylint: disable=redefined-builtin
    """Intercept root-level resolutions and stash metadata on the request.

    Args:
        next (callable): Callable to continue the resolution chain.
        root (object): Parent resolved value. None for top-level fields.
        info (GraphQLResolveInfo): GraphQL resolve info containing operation metadata.
        **kwargs (object): Field arguments.

    Returns:
        object: The result of the resolver.
    """
    if root is not None:
        return next(root, info, **kwargs)

    config = _get_app_settings()
    if not config.get("query_logging_enabled", False):
        return next(root, info, **kwargs)

    # Stash metadata on the request (only for the first root field).
    # For DRF views, info.context is a DRF Request wrapping a WSGIRequest.
    # The Django middleware sees the WSGIRequest, so stash on both.
    request = info.context
    if not hasattr(request, _REQUEST_ATTR):
        meta = {
            "operation_type": info.operation.operation.value,
            "operation_name": PrometheusMiddleware._get_operation_name(info),
            "user": self._get_user(info),
            "config": config,
        }

        if config.get("log_query_body", False):
            meta["query_body"] = _extract_query_body(info)

        if config.get("log_query_variables", False):
            meta["variables"] = _extract_variables(info)

        stash_meta_on_request(request, _REQUEST_ATTR, meta)

    try:
        return next(root, info, **kwargs)
    except Exception as error:
        # Record the error on the stashed metadata so the Django
        # middleware can log it.
        meta = getattr(request, _REQUEST_ATTR, None)
        if meta is not None:
            meta["error"] = error
        raise

nautobot_graphql_observability.metrics

Prometheus metric definitions for GraphQL instrumentation.