Coverage for src / lexigram / contracts / auth / repositories.py: 100%

1"""RepositoryProtocol protocols for auth-domain persistence. 

2 

3Zero-dependency contracts. Any package may depend on these without coupling 

4to a concrete database driver or to each other. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

10 

11if TYPE_CHECKING: 

12 from datetime import datetime 

13 

14__all__ = [ 

15 "APIKeyRepositoryProtocol", 

16 "SessionRepositoryProtocol", 

17] 

18 

19 

20@runtime_checkable 

21class APIKeyRepositoryProtocol(Protocol): 

22 """Persistence contract for API key records. 

23 

24 Implementations are responsible for all SQL/NoSQL I/O. Business-logic 

25 callers (e.g. ``APIKeyManager``) depend only on this protocol, keeping 

26 them decoupled from the underlying data store. 

27 """ 

28 

29 async def insert(self, payload: dict[str, Any]) -> str: 

30 """Persist a new API key record and return the generated key_id. 

31 

32 Args: 

33 payload: Field/value mapping for the new row (name, key_hash, 

34 prefix, user_id, scopes, expires_at, …). 

35 

36 Returns: 

37 The opaque identifier assigned to the record by the store. 

38 """ 

39 ... 

40 

41 async def find_by_prefix(self, prefix: str) -> list[dict[str, Any]]: 

42 """Return all active (non-revoked) keys whose display prefix matches. 

43 

44 Args: 

45 prefix: The short displayable prefix used for fast pre-filtering. 

46 

47 Returns: 

48 List of raw row dicts; empty list when nothing matches. 

49 """ 

50 ... 

51 

52 async def update_last_used(self, key_id: str, ip_address: str | None) -> None: 

53 """Refresh the last-used timestamp and originating IP for a key. 

54 

55 Args: 

56 key_id: Identifier of the key to update. 

57 ip_address: Caller IP, or ``None`` when unavailable. 

58 """ 

59 ... 

60 

61 async def revoke(self, key_id: str) -> None: 

62 """Mark a key as permanently revoked. 

63 

64 Args: 

65 key_id: Identifier of the key to revoke. 

66 """ 

67 ... 

68 

69 async def find_by_user(self, user_id: str) -> list[dict[str, Any]]: 

70 """Return all active (non-revoked) keys for a user. 

71 

72 Args: 

73 user_id: Owner identifier. 

74 

75 Returns: 

76 List of raw row dicts; empty list when nothing matches. 

77 """ 

78 ... 

79 

80 

81@runtime_checkable 

82class SessionRepositoryProtocol(Protocol): 

83 """Persistence contract for user session records. 

84 

85 Decouples session-management business logic from the underlying store, 

86 allowing SQL, Redis, or any other backend to be injected without 

87 altering the service layer. 

88 """ 

89 

90 async def insert(self, payload: dict[str, Any]) -> None: 

91 """Persist a new session record. 

92 

93 Args: 

94 payload: Field/value mapping for the new row (session_id, 

95 admin_id, device_id, ip_address, user_agent, fingerprint, 

96 expires_at, …). 

97 """ 

98 ... 

99 

100 async def find_active(self, session_id: str) -> dict[str, Any] | None: 

101 """Return the row for an active session, or ``None`` if absent/inactive. 

102 

103 Args: 

104 session_id: The opaque session identifier. 

105 

106 Returns: 

107 Raw row dict on success, ``None`` otherwise. 

108 """ 

109 ... 

110 

111 async def find_active_by_user( 

112 self, 

113 user_id: str, 

114 cutoff: datetime, 

115 ) -> list[dict[str, Any]]: 

116 """Return all non-expired, active sessions for a user. 

117 

118 Args: 

119 user_id: The user to query for. 

120 cutoff: Sessions expiring *at or before* this time are excluded. 

121 

122 Returns: 

123 List of raw row dicts ordered by ``last_active_at`` descending. 

124 """ 

125 ... 

126 

127 async def revoke(self, session_id: str) -> None: 

128 """Deactivate a single session. 

129 

130 Args: 

131 session_id: Session to revoke. 

132 """ 

133 ... 

134 

135 async def revoke_all(self, user_id: str) -> None: 

136 """Deactivate every active session owned by a user. 

137 

138 Args: 

139 user_id: Owner whose sessions are to be revoked. 

140 """ 

141 ... 

142 

143 async def update_activity(self, session_id: str, now: datetime) -> None: 

144 """Refresh the ``last_active_at`` timestamp for an active session. 

145 

146 Args: 

147 session_id: Session to touch. 

148 now: Current UTC timestamp to write. 

149 """ 

150 ...