Coverage for src/lexigram/web/middleware/registry.py: 51%

1"""Middleware registry for lexigram-web. 

2 

3Enables extensible adaptation of various middleware types (Starlette, 

4DIScope, Lexigram-native) into the final Starlette middleware stack. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Any, Protocol, cast, runtime_checkable 

10 

11from starlette.middleware import Middleware as StarletteMiddleware 

12 

13 

14@runtime_checkable 

15class _MiddlewareAdapterProtocol(Protocol): 

16 """Protocol for middleware adapters.""" 

17 

18 def can_adapt(self, middleware: Any) -> bool: 

19 """Return True if this adapter can handle the given middleware.""" 

20 ... 

21 

22 def adapt(self, middleware: Any) -> StarletteMiddleware: 

23 """Convert the middleware into a Starlette-compatible Middleware object.""" 

24 ... 

25 

26 

27class _StarletteMiddlewareAdapter: 

28 """Handles native Starlette Middleware objects.""" 

29 

30 def can_adapt(self, middleware: Any) -> bool: 

31 return isinstance(middleware, StarletteMiddleware) 

32 

33 def adapt(self, middleware: Any) -> StarletteMiddleware: 

34 return cast("StarletteMiddleware", middleware) 

35 

36 

37class _TupleMiddlewareAdapter: 

38 """Handles (MiddlewareClass, kwargs_dict) tuples.""" 

39 

40 def can_adapt(self, middleware: Any) -> bool: 

41 return isinstance(middleware, tuple) and len(middleware) >= 1 

42 

43 def adapt(self, middleware: Any) -> StarletteMiddleware: 

44 if len(middleware) == 2 and isinstance(middleware[1], dict): 

45 return StarletteMiddleware(middleware[0], **middleware[1]) 

46 # Positional args (deprecated) 

47 return StarletteMiddleware(middleware[0], *middleware[1:]) 

48 

49 

50class _DIScopeMiddlewareAdapter: 

51 """Handles DIScopeMiddleware instances.""" 

52 

53 def can_adapt(self, middleware: Any) -> bool: 

54 # Check by class name to avoid direct dependency if possible 

55 return type(middleware).__name__ == "DIScopeMiddleware" 

56 

57 def adapt(self, middleware: Any) -> StarletteMiddleware: 

58 return StarletteMiddleware( 

59 cast("Any", type(middleware)), 

60 container=getattr(middleware, "container", None), 

61 ) 

62 

63 

64class _HTTPMiddlewareAdapterBase: 

65 """Handles Starlette BaseHTTPMiddleware instances.""" 

66 

67 def can_adapt(self, middleware: Any) -> bool: 

68 from starlette.middleware.base import BaseHTTPMiddleware 

69 

70 return isinstance(middleware, BaseHTTPMiddleware) 

71 

72 def adapt(self, middleware: Any) -> StarletteMiddleware: 

73 return StarletteMiddleware( 

74 cast("Any", type(middleware)), 

75 container=getattr(middleware, "container", None), 

76 ) 

77 

78 

79class MiddlewareAdapterRegistry: 

80 """Registry for middleware adapters.""" 

81 

82 def __init__(self) -> None: 

83 self._adapters: list[_MiddlewareAdapterProtocol] = [] 

84 self._register_defaults() 

85 

86 def _register_defaults(self) -> None: 

87 # Default order matters: more specific adapters first 

88 self.add_adapter(_StarletteMiddlewareAdapter()) 

89 self.add_adapter(_DIScopeMiddlewareAdapter()) 

90 self.add_adapter(_HTTPMiddlewareAdapterBase()) 

91 self.add_adapter(_TupleMiddlewareAdapter()) 

92 

93 def add_adapter(self, adapter: _MiddlewareAdapterProtocol) -> None: 

94 """Add a custom middleware adapter.""" 

95 self._adapters.insert( 

96 0, 

97 adapter, 

98 ) # LIFO order so custom adapters take precedence 

99 

100 def adapt(self, middleware: Any) -> StarletteMiddleware: 

101 """Adapt a middleware object using registered adapters.""" 

102 for adapter in self._adapters: 

103 if adapter.can_adapt(middleware): 

104 return adapter.adapt(middleware) 

105 

106 # Final fallback: Lexigram Middleware Adapter 

107 from lexigram.web.middleware.adapter import _LexigramMiddlewareAdapter 

108 

109 return StarletteMiddleware(_LexigramMiddlewareAdapter, lexigram_mw=middleware)