Coverage for src/lexigram/auth/di/sub_providers/session_provider.py: 93%

1"""Session provider — registers ``SessionManagerImpl`` in the DI container.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.auth.session.manager import SessionManagerImpl 

8from lexigram.auth.storage.in_memory_stores import InMemorySessionStore 

9from lexigram.contracts.ai.session import SessionManagerProtocol 

10from lexigram.contracts.core import HealthCheckResult, HealthStatus, ProviderPriority 

11from lexigram.di.provider import Provider 

12from lexigram.logging import get_logger 

13 

14if TYPE_CHECKING: 

15 from lexigram.auth.storage.session_store import SessionStore 

16 from lexigram.contracts.core.di import ( 

17 ContainerRegistrarProtocol, 

18 ContainerResolverProtocol, 

19 ) 

20 

21logger = get_logger(__name__) 

22 

23 

24class SessionProvider(Provider): 

25 """Registers :class:`~lexigram.auth.session.manager.SessionManagerImpl` in the container. 

26 

27 Reads ``max_sessions_per_user`` from the supplied ``AuthConfig`` so that 

28 the concurrency limit is driven by application configuration. When no 

29 config is given the session manager defaults to unlimited sessions. 

30 """ 

31 

32 def __init__(self, config: Any = None, **kwargs: Any) -> None: 

33 """Initialise the session provider. 

34 

35 Args: 

36 config: Optional :class:`~lexigram.auth.config.AuthConfig`. Used 

37 to read ``max_sessions_per_user`` and select the session store 

38 backend. When ``None``, sane defaults are used. 

39 **kwargs: Ignored extra keyword arguments (forwarded from bundle). 

40 """ 

41 super().__init__(name="sessions", priority=ProviderPriority.SECURITY) 

42 self._config = config 

43 

44 async def register(self, container: ContainerRegistrarProtocol) -> None: 

45 """Register :class:`SessionManagerImpl` as a singleton. 

46 

47 Args: 

48 container: The DI container registrar. 

49 """ 

50 max_sessions: int | None = None 

51 session_store: SessionStore = InMemorySessionStore() 

52 

53 if self._config is not None: 

54 max_sessions = getattr(self._config, "max_sessions_per_user", None) 

55 

56 session_manager = SessionManagerImpl( 

57 session_store=session_store, 

58 max_sessions_per_user=max_sessions, 

59 ) 

60 container.singleton(SessionManagerImpl, session_manager) 

61 container.singleton(SessionManagerProtocol, session_manager, validate=False) 

62 logger.debug( 

63 "session_provider.registered", 

64 max_sessions_per_user=max_sessions 

65 if max_sessions is not None 

66 else "unlimited", 

67 ) 

68 

69 async def boot(self, container: ContainerResolverProtocol) -> None: 

70 """Boot the session provider. 

71 

72 Args: 

73 container: The read-only container resolver. 

74 """ 

75 logger.info("session_provider.started") 

76 

77 async def shutdown(self) -> None: 

78 """Shut down the session provider.""" 

79 logger.info("session_provider.shutdown") 

80 

81 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

82 """Return the health of the session subsystem. 

83 

84 Args: 

85 timeout: Ignored; present for protocol compatibility. 

86 

87 Returns: 

88 A healthy :class:`~lexigram.contracts.core.HealthCheckResult`. 

89 """ 

90 return HealthCheckResult( 

91 component=self.name, 

92 status=HealthStatus.HEALTHY, 

93 details={ 

94 "max_sessions_per_user": getattr( 

95 self._config, "max_sessions_per_user", None 

96 ) 

97 }, 

98 ) 

99 

100 

101__all__ = ["SessionProvider"]