Coverage for src/lexigram/admin/services/session.py: 0%

1"""Session state management for Lexigram Admin. 

2 

3This module provides request-scoped session state storage via DI container. 

4""" 

5 

6from __future__ import annotations 

7 

8from collections.abc import ItemsView, KeysView, ValuesView 

9import contextvars 

10from typing import Any, ClassVar 

11 

12from lexigram.di.decorators import inject 

13from lexigram.serialization import dumps_str, loads_str 

14 

15# JSONEncoder removed, orjson handles these natively or via default 

16 

17 

18class SessionStateService: 

19 """Request-scoped session state service. 

20 

21 Provides a dict-like interface for storing state within a single request. 

22 Automatically serializes complex types to JSON for persistence. 

23 """ 

24 

25 def __init__(self) -> None: 

26 """Initialize an empty session state.""" 

27 self._data: dict[str, Any] = {} 

28 

29 async def get(self, key: str, default: Any = None) -> Any: 

30 """Get a value from session state (async).""" 

31 return self._data.get(key, default) 

32 

33 async def set(self, key: str, value: Any) -> None: 

34 """Set a value in session state (async).""" 

35 self._data[key] = value 

36 

37 async def delete(self, key: str) -> None: 

38 """Delete a key from session state (async).""" 

39 if key in self._data: 

40 del self._data[key] 

41 else: 

42 raise KeyError(key) 

43 

44 async def clear(self) -> None: 

45 """Clear all session state (async).""" 

46 self._data.clear() 

47 

48 def keys(self) -> KeysView[str]: 

49 """Return all keys in session state.""" 

50 return self._data.keys() 

51 

52 def values(self) -> ValuesView[Any]: 

53 """Return all values in session state.""" 

54 return self._data.values() 

55 

56 def items(self) -> ItemsView[str, Any]: 

57 """Return all key-value pairs in session state.""" 

58 return self._data.items() 

59 

60 def __contains__(self, key: str) -> bool: 

61 """Check if a key exists in session state.""" 

62 return key in self._data 

63 

64 def __getitem__(self, key: str) -> Any: 

65 """Get a value using dict-like syntax.""" 

66 return self._data[key] 

67 

68 def __setitem__(self, key: str, value: Any) -> None: 

69 """Set a value using dict-like syntax.""" 

70 self._data[key] = value 

71 

72 def __delitem__(self, key: str) -> None: 

73 """Delete a value using dict-like syntax.""" 

74 del self._data[key] 

75 

76 def __len__(self) -> int: 

77 """Return the number of items in session state.""" 

78 return len(self._data) 

79 

80 def __repr__(self) -> str: 

81 """Return a string representation of session state.""" 

82 return f"SessionStateService({self._data!r})" 

83 

84 def to_json(self) -> str: 

85 """Serialize session state to JSON.""" 

86 return dumps_str(self._data) 

87 

88 def from_json(self, json_str: str) -> None: 

89 """Load session state from JSON.""" 

90 self._data = loads_str(json_str) 

91 

92 @property 

93 def is_empty(self) -> bool: 

94 """Check if session state is empty.""" 

95 return len(self._data) == 0 

96 

97 

98@inject 

99class ActiveSessionContext: 

100 """Context manager for setting the active session using ContextVars.""" 

101 

102 _session_var: ClassVar[contextvars.ContextVar[SessionStateService | None]] = ( 

103 contextvars.ContextVar("admin_request_session", default=None) 

104 ) 

105 

106 @classmethod 

107 def current(cls) -> SessionStateService | None: 

108 """Get the current active session from context variable.""" 

109 return cls._session_var.get() 

110 

111 def __init__(self, session: SessionStateService): 

112 self.session = session 

113 self._token = None 

114 

115 def __enter__(self) -> SessionStateService: 

116 self._token = ActiveSessionContext._session_var.set(self.session) # type: ignore[assignment] 

117 return self.session 

118 

119 def __exit__(self, exc_type, exc_val, exc_tb) -> Any: 

120 if self._token: 

121 ActiveSessionContext._session_var.reset(self._token)