Coverage for src/lexigram/auth/storage/session_store.py: 100%

16 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-25 12:26 +0800

1"""Session store protocols for abstracting session storage.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

6 

7if TYPE_CHECKING: 

8 from lexigram.auth.models.session import UserSession 

9 

10 

11@runtime_checkable 

12class SessionStore(Protocol): 

13 """Protocol for session storage backends. 

14 

15 Implement this protocol to provide custom session storage 

16 (SQL, Redis, in-memory, etc.). 

17 """ 

18 

19 async def create(self, session: UserSession) -> str: 

20 """Create a new session. 

21 

22 Args: 

23 session: The session to create. 

24 

25 Returns: 

26 The session_id of the created session. 

27 """ 

28 ... 

29 

30 async def get(self, session_id: str) -> UserSession | None: 

31 """Get a session by ID. 

32 

33 Args: 

34 session_id: The session ID to look up. 

35 

36 Returns: 

37 The session if found, None otherwise. 

38 """ 

39 ... 

40 

41 async def update(self, session_id: str, **updates: object) -> None: 

42 """Update session fields. 

43 

44 Args: 

45 session_id: The session ID to update. 

46 **updates: Fields to update. 

47 """ 

48 ... 

49 

50 async def delete(self, session_id: str) -> bool: 

51 """Delete a session. 

52 

53 Args: 

54 session_id: The session ID to delete. 

55 

56 Returns: 

57 True if deleted, False if not found. 

58 """ 

59 ... 

60 

61 async def delete_all_for_user(self, user_id: str) -> int: 

62 """Delete all sessions for a user. 

63 

64 Args: 

65 user_id: The user ID. 

66 

67 Returns: 

68 Number of sessions deleted. 

69 """ 

70 ... 

71 

72 async def list_for_user(self, user_id: str) -> list[UserSession]: 

73 """List all sessions for a user. 

74 

75 Args: 

76 user_id: The user ID. 

77 

78 Returns: 

79 List of active sessions. 

80 """ 

81 ... 

82 

83 

84@runtime_checkable 

85class MFAStore(Protocol): 

86 """Protocol for MFA storage backends. 

87 

88 Implement this protocol to provide custom MFA storage 

89 (SQL, in-memory, etc.). 

90 """ 

91 

92 async def get(self, user_id: str) -> dict[str, Any] | None: 

93 """Get MFA config for a user. 

94 

95 Args: 

96 user_id: The user ID. 

97 

98 Returns: 

99 MFA data if found, None otherwise. 

100 """ 

101 ... 

102 

103 async def save(self, mfa_data: dict[str, object]) -> None: 

104 """Save MFA config for a user. 

105 

106 Args: 

107 mfa_data: MFA data to save. 

108 """ 

109 ... 

110 

111 async def delete(self, user_id: str) -> bool: 

112 """Delete MFA config for a user. 

113 

114 Args: 

115 user_id: The user ID. 

116 

117 Returns: 

118 True if deleted. 

119 """ 

120 ... 

121 

122 

123__all__ = [ 

124 "MFAStore", 

125 "SessionStore", 

126]