Coverage for src/lexigram/web/filters/pipeline.py: 50%
58 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Filter pipeline execution for exception handling.
3Provides production-grade exception filter pipeline with DI integration.
4"""
6from __future__ import annotations
8import traceback
9from typing import Any, cast
11from lexigram.contracts.web.protocols import ExceptionFilterProtocol
12from lexigram.logging import get_logger
13from lexigram.web.transport.responses import JSONResponse, Response
15logger = get_logger(__name__)
18class FilterPipeline:
19 """Production-grade exception filter pipeline with DI integration.
21 Provides:
22 - add_filter() with O(1) index by exception type
23 - remove_filter() to remove filters by type
24 - MRO-based exception matching for subclass support
25 - Fallback to default 500 response
26 """
28 def __init__(
29 self, filters: list[ExceptionFilterProtocol] | None = None, debug: bool = False
30 ):
31 """Initialize the filter pipeline.
33 Args:
34 filters: Optional list of initial filters.
35 debug: When ``True``, unhandled exceptions render an HTML debug
36 page for browser clients instead of a generic JSON 500.
37 """
38 self._filters: list[ExceptionFilterProtocol] = list(filters) if filters else []
39 self._debug = debug
40 # Type index for O(1) lookup by exact exception type
41 self._type_index: dict[type[Exception], ExceptionFilterProtocol] = {}
42 # Rebuild index on init
43 self._rebuild_index()
45 def _rebuild_index(self) -> None:
46 """Rebuild the type index from filters."""
47 self._type_index.clear()
48 for f in self._filters:
49 if hasattr(f, "exception_type"):
50 self._type_index[f.exception_type] = f
52 @property
53 def filters(self) -> list[ExceptionFilterProtocol]:
54 """Get all registered filters."""
55 return list(self._filters)
57 def add_filter(self, filter_instance: ExceptionFilterProtocol) -> None:
58 """Add filter and index by exception type for O(1) lookup.
60 Args:
61 filter_instance: The filter to add.
62 """
63 if filter_instance not in self._filters:
64 self._filters.append(filter_instance)
66 # Index by exception type if available
67 if hasattr(filter_instance, "exception_type"):
68 self._type_index[filter_instance.exception_type] = filter_instance
70 def remove_filter(self, filter_type: type[ExceptionFilterProtocol]) -> None:
71 """Remove filter by type.
73 Args:
74 filter_type: The type of filter to remove.
75 """
76 self._filters = [
77 f
78 for f in self._filters
79 if not (isinstance(f, filter_type) or type(f) is filter_type)
80 ]
81 # Rebuild index
82 self._rebuild_index()
84 def clear(self) -> None:
85 """Clear all filters.
87 Use in tests only to prevent test pollution between test cases.
88 """
89 self._filters.clear()
90 self._rebuild_index()
92 async def handle(self, exc: Exception, request: Any) -> Response:
93 """Find matching filter via MRO walk for exception subclass support.
95 Args:
96 exc: The exception that was raised.
97 request: The HTTP request.
99 Returns:
100 Response from the matching filter, or default 500.
101 """
102 # Try exact type match first
103 exc_type = type(exc)
104 if exc_type in self._type_index:
105 return cast(
106 "Response",
107 await cast("Any", self._type_index[exc_type]).handle(exc, request),
108 )
110 # Walk MRO for subclass matching
111 for klass in exc_type.__mro__:
112 if klass in self._type_index:
113 return cast(
114 "Response",
115 await cast("Any", self._type_index[klass]).handle(exc, request),
116 )
118 # Try filter-level matching (last added runs first)
119 for f in reversed(self._filters):
120 if f.can_handle(exc):
121 return cast("Response", await cast("Any", f).handle(exc, request))
123 # Default: structured 500 response (HTML in debug mode for browsers)
124 logger.error(
125 "unhandled_exception_in_filter_pipeline",
126 error=str(exc),
127 error_type=type(exc).__name__,
128 )
130 if self._debug:
131 from lexigram.web.errors.html_error_renderer import DebugHtmlErrorRenderer
133 renderer = DebugHtmlErrorRenderer()
134 if renderer.should_render(request):
135 return renderer.render(exc, request, status_code=500) # type: ignore[return-value]
137 return JSONResponse(
138 content={
139 "success": False,
140 "error": {
141 "type": "internal_server_error",
142 "message": str(exc),
143 "exception_type": type(exc).__name__,
144 "traceback": traceback.format_exc(),
145 },
146 },
147 status_code=500,
148 )
150 return JSONResponse(
151 content={
152 "success": False,
153 "error": {
154 "type": "internal_server_error",
155 "message": "An unexpected error occurred",
156 },
157 },
158 status_code=500,
159 )
162# Global filter pipeline singleton — registered in DI container during boot.
163# Mutation controlled via add_global_filter()/remove_global_filter().
164filter_pipeline = FilterPipeline()
167def get_filter_pipeline(context: Any | None = None) -> FilterPipeline | None:
168 """Get the global filter pipeline, or None if no filters are registered.
170 Returns None when the global pipeline is empty so that callers can
171 re-raise and let upstream handlers (e.g. ExceptionFilterRegistry) handle
172 the exception instead.
173 """
174 return filter_pipeline if filter_pipeline.filters else None
177def add_global_filter(filter_instance: ExceptionFilterProtocol) -> None:
178 """Add a filter to the global pipeline.
180 Args:
181 filter_instance: The filter to add.
182 """
183 filter_pipeline.add_filter(filter_instance)
186def remove_global_filter(filter_type: type[ExceptionFilterProtocol]) -> None:
187 """Remove a filter from the global pipeline.
189 Args:
190 filter_type: The type of filter to remove.
191 """
192 filter_pipeline.remove_filter(filter_type)
195__all__ = [
196 "FilterPipeline",
197 "add_global_filter",
198 "filter_pipeline",
199 "get_filter_pipeline",
200 "remove_global_filter",
201]