Coverage for src/lexigram/web/middleware/di_scope.py: 18%

1"""DI Scope Middleware for request-scoped dependency injection. 

2 

3Ensures DI container is properly scoped per request to prevent 

4data leaks and memory issues. 

5 

6This middleware: 

71. Creates a child container for each request 

82. Attaches it to request.state.container 

93. Properly disposes of scoped resources after the request 

10""" 

11 

12from __future__ import annotations 

13 

14from typing import TYPE_CHECKING, Any, cast 

15import uuid 

16 

17from starlette.requests import Request 

18 

19from lexigram.logging import get_logger 

20 

21logger = get_logger(__name__) 

22 

23if TYPE_CHECKING: 

24 from starlette.types import ASGIApp, Receive, Scope, Send 

25 

26 from lexigram.di.container import Container 

27 

28 

29class DIScopeMiddleware: 

30 """ 

31 Pure ASGI middleware that creates a request-scoped DI container. 

32 

33 Creates a child container for each request to ensure proper 

34 isolation of request-scoped services (like UnitOfWork, RequestContext). 

35 """ 

36 

37 def __init__( 

38 self, 

39 app: ASGIApp, 

40 container: Container | None = None, 

41 scoped_services: list[type] | None = None, 

42 ): 

43 """ 

44 Initialize the DI scope middleware. 

45 

46 Args: 

47 app: The ASGI application 

48 container: The root DI container 

49 scoped_services: Optional list of service types to register as scoped 

50 """ 

51 self.app = app 

52 self.container = container 

53 self.scoped_services = scoped_services or [] 

54 

55 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

56 """Pure ASGI middleware implementation - no task creation overhead.""" 

57 

58 if scope["type"] != "http": 

59 # Pass through non-HTTP requests (WebSocket, lifespan, etc.) 

60 await self.app(scope, receive, send) 

61 return 

62 

63 # Extract request info from ASGI scope 

64 headers = dict(scope.get("headers", [])) 

65 request_id = headers.get(b"x-request-id", str(uuid.uuid4()).encode()).decode() 

66 

67 # Create request-scoped container 

68 request_container: Any = None 

69 service_scope: Any = None 

70 if self.container is not None: 

71 try: 

72 # Create a scope for this request using create_scope() 

73 if hasattr(self.container, "create_scope"): 

74 service_scope = self.container.create_scope() 

75 request_container = service_scope 

76 elif hasattr(self.container, "child"): 

77 # Fallback to child() if create_scope() not available 

78 request_container = self.container.child() 

79 else: 

80 raise RuntimeError( 

81 "Container has no create_scope() or child() method — cannot create request scope" 

82 ) 

83 

84 # Register scoped services in the scoped container 

85 for service_type in self.scoped_services: 

86 if hasattr(request_container, "register_scoped"): 

87 request_container.register_scoped(service_type, service_type) 

88 elif hasattr(request_container, "scoped"): 

89 request_container.scoped(service_type) 

90 except (RuntimeError, LookupError) as exc: 

91 logger.error("di_scope_creation_failed", error=str(exc)) 

92 raise RuntimeError("Failed to create request scope") from exc 

93 

94 try: 

95 # Create a minimal Request object for DI scope 

96 request = Request(scope, receive) 

97 

98 # Attach scoped container and request to request state 

99 request.state.container = request_container 

100 request.state.root_container = self.container 

101 request.state.request_id = request_id 

102 

103 # Use service_scope for disposal check later 

104 

105 # Also make available via the container if possible 

106 if request_container is not None and hasattr( 

107 request_container, 

108 "__setitem__", 

109 ): 

110 request_container["request"] = request 

111 request_container["request_id"] = request_id 

112 

113 # Process request with scoped container 

114 await self.app(scope, receive, send) 

115 

116 # Scope exits here -> automatic cleanup of scoped resources 

117 # (database sessions, connections, etc.) 

118 if service_scope is not None and hasattr(service_scope, "dispose"): 

119 try: 

120 await cast("Any", service_scope).dispose() 

121 except (RuntimeError, AttributeError) as exc: 

122 logger.debug( 

123 "scope_dispose_failed", error=str(exc) 

124 ) # Best effort cleanup 

125 elif request_container is not None and hasattr( 

126 request_container, 

127 "dispose", 

128 ): 

129 try: 

130 await cast("Any", request_container).dispose() 

131 except (RuntimeError, AttributeError) as exc: 

132 logger.debug( 

133 "container_dispose_failed", error=str(exc) 

134 ) # Best effort cleanup 

135 

136 finally: 

137 pass # Context vars are managed by RequestContextMiddleware 

138 

139 

140__all__ = ["DIScopeMiddleware"]