Coverage for src/lexigram/web/integrations/setup.py: 27%

1"""ASGI lifespan and application setup helpers for the web provider.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import AsyncIterator 

6from contextlib import asynccontextmanager 

7import inspect 

8from typing import TYPE_CHECKING 

9 

10from lexigram.logging import get_logger 

11from lexigram.web.hooks import WebServerStartedHook, WebServerStoppedHook 

12 

13if TYPE_CHECKING: 

14 from starlette.applications import Starlette 

15 

16logger = get_logger(__name__) 

17 

18 

19async def _emit_action(app: Starlette, hook_name: str, payload: object) -> None: 

20 """Emit a lifecycle action hook when the app is wired with a registry.""" 

21 hooks = getattr(app.state, "hook_registry", None) 

22 if hooks is None: 

23 return 

24 

25 await hooks.call_action(hook_name, payload=payload) 

26 

27 

28@asynccontextmanager 

29async def lifespan(app: Starlette) -> AsyncIterator[None]: 

30 """ASGI lifespan manager for the Starlette application. 

31 

32 Handles startup diagnostics and graceful shutdown of shared resources 

33 attached to ``app.state`` (database pool, Redis client, etc.). 

34 """ 

35 logger.info("Starting web application lifespan") 

36 

37 # Initialize background_tasks if not already present 

38 if not hasattr(app.state, "background_tasks"): 

39 app.state.background_tasks = [] 

40 

41 # Diagnostics for tests that expect debug logs on missing resources 

42 if not hasattr(app.state, "db_pool"): 

43 logger.debug( 

44 "Database pool not found in app state, skipping cleanup registration", 

45 ) 

46 

47 await _emit_action(app, "server.started", WebServerStartedHook()) 

48 

49 yield 

50 

51 logger.info("Shutting down web application lifespan") 

52 

53 # 1. Cleanup DB Pool 

54 if hasattr(app.state, "db_pool"): 

55 try: 

56 pool = app.state.db_pool 

57 if hasattr(pool, "close"): 

58 res = pool.close() 

59 if hasattr(res, "__await__") or inspect.isawaitable(res): 

60 await res 

61 except Exception: # noqa: BLE001 — best-effort DB pool cleanup, must not crash shutdown 

62 logger.exception("Error closing DB pool") 

63 

64 # 2. Cleanup Redis Client 

65 if hasattr(app.state, "redis_client"): 

66 try: 

67 client = app.state.redis_client 

68 if hasattr(client, "close"): 

69 res = client.close() 

70 if hasattr(res, "__await__") or inspect.isawaitable(res): 

71 await res 

72 except Exception: # noqa: BLE001 — best-effort Redis client cleanup, must not crash shutdown 

73 logger.exception("Error closing Redis client") 

74 

75 await _emit_action(app, "server.stopped", WebServerStoppedHook()) 

76 

77 logger.debug("Web application lifespan ended") 

78 

79 

80__all__ = ["lifespan"]