Coverage for src/lexigram/auth/storage/apikey_sql.py: 50%

1"""SQL-backed implementation of APIKeyRepositoryProtocol. 

2 

3This is the only module in ``lexigram-auth`` allowed to issue raw SQL for 

4API key persistence. All callers depend on the ``APIKeyRepositoryProtocol`` protocol 

5from ``lexigram-contracts`` — never on this concrete class directly. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, Any 

11 

12from lexigram.logging import get_logger 

13 

14if TYPE_CHECKING: 

15 from lexigram.contracts import DatabaseProviderProtocol 

16 

17logger = get_logger(__name__) 

18 

19 

20class APIKeySqlRepository: 

21 """SQL-backed repository for API key persistence. 

22 

23 Wraps a ``DatabaseProviderProtocol`` and implements the ``APIKeyRepositoryProtocol`` 

24 protocol so it can be injected wherever that protocol is required. 

25 

26 All raw SQL lives here; nothing outside this module constructs queries for 

27 the ``api_keys`` table. 

28 """ 

29 

30 _TABLE = "api_keys" 

31 

32 def __init__(self, db_provider: DatabaseProviderProtocol) -> None: 

33 """Initialise with a resolved database provider. 

34 

35 Args: 

36 db_provider: Framework database provider that exposes 

37 ``execute_insert``, ``execute_query``, and ``execute_sql``. 

38 """ 

39 self._db = db_provider 

40 

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

42 """Persist a new API key row and return the generated key_id. 

43 

44 Args: 

45 payload: Field/value mapping (name, key_hash, prefix, user_id, 

46 scopes, expires_at, …). 

47 

48 Returns: 

49 The opaque key identifier returned by the store. 

50 """ 

51 result = await self._db.execute_insert(self._TABLE, payload) 

52 return str(result) 

53 

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

55 """Return all active (non-revoked) rows matching the display prefix. 

56 

57 Args: 

58 prefix: Short display prefix used for fast pre-filtering. 

59 

60 Returns: 

61 List of row dicts; empty when nothing matches. 

62 """ 

63 sql = f"SELECT * FROM {self._TABLE} WHERE prefix = ? AND revoked_at IS NULL" # noqa: S608 — table name is constant class attr "api_keys", never user input 

64 result = await self._db.execute_query(sql, [prefix]) 

65 return list(result.rows) 

66 

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

68 """Refresh the ``last_used_at`` timestamp and originating IP. 

69 

70 Args: 

71 key_id: Target key identifier. 

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

73 """ 

74 sql = ( 

75 f"UPDATE {self._TABLE} " # noqa: S608 — table name is constant class attr "api_keys", never user input 

76 "SET last_used_at = NOW(), last_used_ip = ? " 

77 "WHERE id = ?" 

78 ) 

79 await self._db.execute(sql, [ip_address, key_id]) 

80 

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

82 """Mark a key as permanently revoked. 

83 

84 Args: 

85 key_id: Target key identifier. 

86 """ 

87 sql = ( 

88 f"UPDATE {self._TABLE} " # noqa: S608 — table name is constant class attr "api_keys", never user input 

89 "SET revoked_at = NOW(), updated_at = NOW() " 

90 "WHERE id = ?" 

91 ) 

92 await self._db.execute(sql, [key_id]) 

93 

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

95 """Return all active (non-revoked) keys owned by a user. 

96 

97 Args: 

98 user_id: Owner identifier. 

99 

100 Returns: 

101 List of row dicts; empty when nothing matches. 

102 """ 

103 sql = f"SELECT * FROM {self._TABLE} WHERE user_id = ? AND revoked_at IS NULL" # noqa: S608 — table name is constant class attr "api_keys", never user input 

104 result = await self._db.execute_query(sql, [user_id]) 

105 return list(result.rows) 

106 

107 

108__all__ = [ 

109 "APIKeySqlRepository", 

110 "logger", 

111]