Coverage for src/lexigram/web/middleware/hooks.py: 36%

1"""Pure ASGI middleware for canonical web hook emission.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import MutableMapping 

6from typing import TYPE_CHECKING, Any, cast 

7 

8from lexigram.contracts.core import HookRegistryProtocol 

9from lexigram.web.hooks import WebRequestReceivedHook, WebResponsePreparedHook 

10 

11if TYPE_CHECKING: 

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

13 

14 

15class WebHooksMiddleware: 

16 """Emit request/response lifecycle hooks around HTTP ASGI traffic.""" 

17 

18 def __init__( 

19 self, 

20 app: ASGIApp, 

21 hooks: HookRegistryProtocol | None = None, 

22 ) -> None: 

23 self.app = app 

24 self._hooks = hooks 

25 

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

27 """Emit canonical hooks for HTTP requests and prepared responses.""" 

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

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

30 return 

31 

32 path = cast("str", scope.get("path", "")) 

33 method = cast("str", scope.get("method", "")) 

34 await self._emit_action( 

35 "request.received", 

36 WebRequestReceivedHook(path=path, method=method), 

37 ) 

38 

39 response_started = False 

40 

41 async def send_with_hooks(message: MutableMapping[str, Any]) -> None: 

42 nonlocal response_started 

43 if message["type"] == "http.response.start" and not response_started: 

44 response_started = True 

45 await self._emit_action( 

46 "response.prepared", 

47 WebResponsePreparedHook( 

48 path=path, 

49 status_code=cast("int", message.get("status", 200)), 

50 ), 

51 ) 

52 await send(message) 

53 

54 await self.app(scope, receive, send_with_hooks) 

55 

56 async def _emit_action(self, hook_name: str, payload: object) -> None: 

57 """Emit a hook action when a registry is available.""" 

58 if self._hooks is None: 

59 return 

60 

61 await self._hooks.call_action(hook_name, payload=payload) 

62 

63 

64__all__ = ["WebHooksMiddleware"]