Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-governance/src/lexigram/ai/governance/relay_logs/persistence.py: 58%

1"""SQL-backed relay request-log store. 

2 

3Persists redaction-safe dispatch entries through 

4:class:`~lexigram.contracts.data.DatabaseProviderProtocol` using only 

5generic ``execute``/``execute_query`` SQL so the store works on any 

6backend. Writes are insert-only and idempotent: ``request_id`` is the 

7primary key and a duplicate append is ignored, so retries and 

8concurrent writes never double-record a dispatch. 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import TYPE_CHECKING 

14 

15from lexigram.contracts.ai.relay import ( 

16 RelayRequestLogEntry, 

17 RelayRequestLogStoreProtocol, 

18) 

19 

20if TYPE_CHECKING: 

21 from lexigram.contracts.data import DatabaseProviderProtocol 

22 

23__all__ = ["SqlRelayRequestLogStore"] 

24 

25_CREATE_LOG_TABLE = """ 

26CREATE TABLE IF NOT EXISTS ai_relay_request_logs ( 

27 request_id TEXT NOT NULL PRIMARY KEY, 

28 user_id TEXT NOT NULL, 

29 token_id TEXT NOT NULL, 

30 endpoint_kind TEXT NOT NULL, 

31 model TEXT NOT NULL, 

32 channel_name TEXT NOT NULL DEFAULT '', 

33 status TEXT NOT NULL, 

34 created_at TEXT NOT NULL, 

35 prompt_tokens INTEGER NOT NULL DEFAULT 0, 

36 completion_tokens INTEGER NOT NULL DEFAULT 0, 

37 cost TEXT NOT NULL DEFAULT '0', 

38 latency_ms INTEGER NOT NULL DEFAULT 0, 

39 error_code TEXT NOT NULL DEFAULT '' 

40) 

41""" 

42 

43_CREATE_LOG_INDEX = """ 

44CREATE INDEX IF NOT EXISTS idx_ai_relay_request_logs_scope 

45 ON ai_relay_request_logs (user_id, model, created_at) 

46""" 

47 

48_INSERT_LOG = ( 

49 "INSERT OR IGNORE INTO ai_relay_request_logs " 

50 "(request_id, user_id, token_id, endpoint_kind, model, channel_name, " 

51 "status, created_at, prompt_tokens, completion_tokens, cost, latency_ms, " 

52 "error_code) " 

53 "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" 

54) 

55 

56 

57class SqlRelayRequestLogStore(RelayRequestLogStoreProtocol): 

58 """SQL-backed request-log sink. 

59 

60 Creates the ``ai_relay_request_logs`` table lazily on first use and 

61 appends entries insert-only; a duplicate ``request_id`` is ignored. 

62 

63 Args: 

64 db: A connected 

65 :class:`~lexigram.contracts.data.DatabaseProviderProtocol` 

66 resolved from the DI container. 

67 """ 

68 

69 def __init__(self, db: DatabaseProviderProtocol) -> None: 

70 self._db = db 

71 self._initialised = False 

72 

73 async def _ensure_tables(self) -> None: 

74 """Create the storage schema once, on first use.""" 

75 if not self._initialised: 

76 await self._db.execute(_CREATE_LOG_TABLE) 

77 await self._db.execute(_CREATE_LOG_INDEX) 

78 self._initialised = True 

79 

80 async def append(self, entry: RelayRequestLogEntry) -> None: 

81 """Persist *entry*, ignoring a duplicate ``request_id``. 

82 

83 Args: 

84 entry: The redaction-safe dispatch entry to persist. 

85 """ 

86 await self._ensure_tables() 

87 await self._db.execute( 

88 _INSERT_LOG, 

89 [ 

90 entry.request_id, 

91 entry.user_id, 

92 entry.token_id, 

93 entry.endpoint_kind, 

94 entry.model, 

95 entry.channel_name, 

96 entry.status, 

97 entry.created_at.isoformat(), 

98 entry.prompt_tokens, 

99 entry.completion_tokens, 

100 entry.cost, 

101 entry.latency_ms, 

102 entry.error_code, 

103 ], 

104 )