Coverage for src/lexigram/graphql/core/middleware.py: 100%
96 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""GraphQL execution middleware pipeline.
3This module provides middleware support for GraphQL execution,
4enabling cross-cutting concerns like logging, auth, and metrics.
6**Why GraphQL middleware is intentionally distinct from HTTP middleware**
7------------------------------------------------------------------------
9The HTTP/ASGI middleware contract (``lexigram.contracts.web.middleware.Middleware``)
10operates at the ASGI transport layer — it receives raw ``scope``/``receive``/``send``
11arguments and wraps the full HTTP connection.
13GraphQL middleware here operates at the **GraphQL execution layer** — it wraps
14individual *field resolution or operation execution* callbacks. The concerns and
15signatures are fundamentally different:
17- HTTP middleware: ``(scope, receive, send) → None``
18- GraphQL middleware: ``(context, next_handler) → response``
20Aligning them into a single protocol would force an artificial abstraction and
21couple ``lexigram-graphql`` to ASGI internals. They are therefore intentionally
22separate.
24If you need to apply cross-cutting HTTP logic to the GraphQL endpoint (e.g.,
25auth, rate-limiting, request-id injection), do so via the HTTP middleware stack
26configured in ``WebProvider``. Use ``GraphQLMiddleware`` only for logic that
27needs access to the parsed GraphQL operation (query, variables, context).
28"""
30from __future__ import annotations
32from abc import ABC
33from dataclasses import dataclass
34from typing import TYPE_CHECKING, Any, Protocol, TypeVar
36from lexigram.logging import get_logger
38if TYPE_CHECKING:
39 from collections.abc import Callable
41logger = get_logger(__name__)
43# Type variable for generic middleware
44T = TypeVar("T")
47@dataclass
48class MiddlewareResult:
49 """Result of middleware processing.
51 Attributes:
52 response: The GraphQL response (if middleware handled it).
53 skipped: Whether the middleware skipped processing.
54 error: Any error that occurred.
55 """
57 response: Any = None
58 skipped: bool = False
59 error: Exception | None = None
62class GraphQLMiddleware(Protocol):
63 """Protocol for GraphQL middleware.
65 Middleware can inspect, modify, or short-circuit GraphQL requests.
66 """
68 async def process(
69 self,
70 context: Any,
71 next_handler: Callable[..., Any],
72 ) -> Any:
73 """Process a GraphQL operation.
75 Args:
76 context: The execution context.
77 next_handler: The next handler in the chain.
79 Returns:
80 GraphQL response.
81 """
82 ...
85class AbstractMiddleware(ABC):
86 """Abstract base class for GraphQL middleware.
88 Provides common functionality for middleware implementations.
89 """
91 def __init__(self, enabled: bool = True):
92 """Initialize the middleware.
94 Args:
95 enabled: Whether this middleware is enabled.
96 """
97 self._enabled = enabled
99 @property
100 def enabled(self) -> bool:
101 """Check if middleware is enabled."""
102 return self._enabled
104 async def process(
105 self,
106 context: Any,
107 next_handler: Callable[..., Any],
108 ) -> Any:
109 """Process a GraphQL operation.
111 Default implementation just calls the next handler.
112 Subclasses should override to add custom logic.
113 """
114 return await next_handler(context)
117class LoggingMiddleware(AbstractMiddleware):
118 """Middleware for logging GraphQL operations.
120 Logs request and response information.
121 """
123 def __init__(
124 self,
125 enabled: bool = True,
126 log_variables: bool = False,
127 ):
128 """Initialize the logging middleware.
130 Args:
131 enabled: Whether logging is enabled.
132 log_variables: Whether to log query variables.
133 """
134 super().__init__(enabled)
135 self._log_variables = log_variables
137 async def process(
138 self,
139 context: Any,
140 next_handler: Callable[..., Any],
141 ) -> Any:
142 """Process and log GraphQL operation."""
143 if not self._enabled:
144 return await next_handler(context)
146 request = getattr(context, "request", None)
147 query = getattr(request, "query", "") if request else "unknown"
149 # Log request
150 logger.debug(
151 "GraphQL request: %s",
152 query[:100] + "..." if len(query) > 100 else query,
153 )
155 # Execute
156 try:
157 response = await next_handler(context)
158 logger.debug("GraphQL response: success")
159 return response
160 except Exception as _gql_err: # noqa: BLE001 — middleware must intercept any exception for logging before re-raising
161 logger.exception("GraphQL request failed")
162 raise
165class AuthMiddleware(AbstractMiddleware):
166 """Middleware for authentication/authorization.
168 Validates user authentication before execution.
169 """
171 def __init__(
172 self,
173 enabled: bool = True,
174 require_auth: bool = False,
175 ):
176 """Initialize the auth middleware.
178 Args:
179 enabled: Whether auth is enabled.
180 require_auth: Whether authentication is required.
181 """
182 super().__init__(enabled)
183 self._require_auth = require_auth
185 async def process(
186 self,
187 context: Any,
188 next_handler: Callable[..., Any],
189 ) -> Any:
190 """Process authentication check."""
191 if not self._enabled:
192 return await next_handler(context)
194 # Check for user in context
195 user = getattr(context, "user", None)
197 if self._require_auth and user is None:
198 from lexigram.graphql.exceptions import AuthenticationError
200 raise AuthenticationError("Authentication required")
202 return await next_handler(context)
205class MetricsMiddleware(AbstractMiddleware):
206 """Middleware for collecting execution metrics.
208 Tracks execution time and other metrics.
209 """
211 def __init__(self, enabled: bool = True):
212 """Initialize the metrics middleware."""
213 super().__init__(enabled)
214 self._metrics: dict[str, Any] = {}
216 async def process(
217 self,
218 context: Any,
219 next_handler: Callable[..., Any],
220 ) -> Any:
221 """Process and collect metrics."""
222 if not self._enabled:
223 return await next_handler(context)
225 import time
227 start = time.perf_counter()
229 try:
230 response = await next_handler(context)
231 duration = time.perf_counter() - start
233 # Record metrics
234 self._metrics["total_requests"] = self._metrics.get("total_requests", 0) + 1
235 self._metrics["total_time"] = self._metrics.get("total_time", 0) + duration
237 return response
238 except Exception as _metrics_err: # noqa: BLE001 — metrics middleware must capture any error to track error count before re-raising
239 self._metrics["total_errors"] = self._metrics.get("total_errors", 0) + 1
240 raise
242 def get_metrics(self) -> dict[str, Any]:
243 """Get collected metrics."""
244 return self._metrics.copy()
247class MiddlewarePipeline:
248 """Chain of middleware executed around GraphQL operations.
250 Example:
251 ```python
252 pipeline = MiddlewarePipeline([
253 LoggingMiddleware(),
254 AuthMiddleware(),
255 MetricsMiddleware(),
256 ])
258 response = await pipeline.execute(context, handler)
259 ```
260 """
262 def __init__(self, middlewares: list[AbstractMiddleware] | None = None):
263 """Initialize the pipeline.
265 Args:
266 middlewares: List of middleware to apply.
267 """
268 self._middlewares = middlewares or []
270 def add(self, middleware: AbstractMiddleware) -> MiddlewarePipeline:
271 """Add a middleware to the pipeline.
273 Args:
274 middleware: The middleware to add.
276 Returns:
277 Self for chaining.
278 """
279 self._middlewares.append(middleware)
280 return self
282 async def execute(
283 self,
284 context: Any,
285 handler: Callable[..., Any],
286 ) -> Any:
287 """Execute the middleware pipeline.
289 Args:
290 context: Execution context.
291 handler: Final handler to call.
293 Returns:
294 GraphQL response.
295 """
296 # Build the chain of handlers
297 next_handler = handler
299 # Reverse order so first middleware runs first
300 for middleware in reversed(self._middlewares):
301 if not middleware.enabled:
302 continue
304 current_middleware = middleware
305 next_handler = lambda ctx, next=next_handler: current_middleware.process(
306 ctx, next
307 )
309 # Execute the chain
310 return await next_handler(context)
313def create_middleware_pipeline(
314 config: dict[str, bool] | None = None,
315) -> MiddlewarePipeline:
316 """Create a middleware pipeline from configuration.
318 Args:
319 config: Configuration dict with middleware settings.
321 Returns:
322 Configured MiddlewarePipeline.
323 """
324 config = config or {}
326 pipeline = MiddlewarePipeline()
328 if config.get("logging", True):
329 pipeline.add(LoggingMiddleware())
331 if config.get("auth", False):
332 pipeline.add(AuthMiddleware(require_auth=config.get("require_auth", False)))
334 if config.get("metrics", True):
335 pipeline.add(MetricsMiddleware())
337 return pipeline
340__all__ = [
341 "AbstractMiddleware",
342 "AuthMiddleware",
343 "GraphQLMiddleware",
344 "LoggingMiddleware",
345 "MetricsMiddleware",
346 "MiddlewarePipeline",
347 "MiddlewareResult",
348 "create_middleware_pipeline",
349]