Coverage for packages / core / storage / audit.py: 38%

1"""Audit logging for data changes and corrections. 

2 

3Provides the AuditLogger class for recording and querying data modifications 

4across the system, enabling traceability of data corrections. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Any 

10 

11from sqlalchemy import desc 

12from sqlalchemy.orm import Session 

13 

14from packages.core.common.logging import get_logger 

15from packages.core.storage.models import AuditLog 

16 

17logger = get_logger(__name__) 

18 

19 

20class AuditLogger: 

21 """Logger for recording and querying data changes. 

22 

23 Usage:: 

24 

25 audit = AuditLogger() 

26 audit.log_change( 

27 session, table_name="starters", record_id="42", 

28 action="CORRECT", old_values={"placing": 3}, new_values={"placing": 2}, 

29 changed_by="admin", change_reason="Manual correction from video review", 

30 ) 

31 

32 All methods are safe to call — failures are logged but never raised, 

33 ensuring audit logging never blocks the primary operation. 

34 """ 

35 

36 @staticmethod 

37 def log_change( 

38 session: Session, 

39 table_name: str, 

40 record_id: str, 

41 action: str, 

42 old_values: dict[str, Any] | None = None, 

43 new_values: dict[str, Any] | None = None, 

44 changed_by: str | None = None, 

45 change_reason: str | None = None, 

46 ) -> AuditLog | None: 

47 """Record a data change in the audit log. 

48 

49 Args: 

50 session: Database session. 

51 table_name: Name of the table that was changed. 

52 record_id: Primary key value of the changed record (stringified). 

53 action: Type of change — ``INSERT``, ``UPDATE``, ``DELETE``, or ``CORRECT``. 

54 old_values: Snapshot of values before the change (optional). 

55 new_values: Snapshot of values after the change (optional). 

56 changed_by: Identifier of the user/system that made the change (optional). 

57 change_reason: Human-readable reason for the change (optional). 

58 

59 Returns: 

60 The created AuditLog entry, or ``None`` if logging failed. 

61 """ 

62 action_upper = action.upper().strip() 

63 if action_upper not in ("INSERT", "UPDATE", "DELETE", "CORRECT"): 

64 logger.warning( 

65 "Invalid audit action '%s' — must be INSERT, UPDATE, DELETE, or CORRECT", 

66 action, 

67 ) 

68 return None 

69 

70 try: 

71 entry = AuditLog( 

72 table_name=table_name, 

73 record_id=str(record_id), 

74 action=action_upper, 

75 old_values=old_values, 

76 new_values=new_values, 

77 changed_by=changed_by, 

78 change_reason=change_reason, 

79 ) 

80 session.add(entry) 

81 session.flush() 

82 logger.debug( 

83 "Audit log entry created", 

84 extra={ 

85 "table_name": table_name, 

86 "record_id": record_id, 

87 "action": action_upper, 

88 }, 

89 ) 

90 return entry 

91 except Exception: 

92 logger.exception( 

93 "Failed to create audit log entry", 

94 extra={ 

95 "table_name": table_name, 

96 "record_id": record_id, 

97 "action": action_upper, 

98 }, 

99 ) 

100 return None 

101 

102 @staticmethod 

103 def get_changes_for_record( 

104 session: Session, 

105 table_name: str, 

106 record_id: str, 

107 ) -> list[AuditLog]: 

108 """Retrieve all audit log entries for a specific record. 

109 

110 Args: 

111 session: Database session. 

112 table_name: Name of the table. 

113 record_id: Primary key value of the record (stringified). 

114 

115 Returns: 

116 List of AuditLog entries, newest first. 

117 """ 

118 try: 

119 return ( 

120 session.query(AuditLog) 

121 .filter( 

122 AuditLog.table_name == table_name, 

123 AuditLog.record_id == str(record_id), 

124 ) 

125 .order_by(desc(AuditLog.created_at)) 

126 .all() 

127 ) 

128 except Exception: 

129 logger.exception( 

130 "Failed to query audit log for record", 

131 extra={"table_name": table_name, "record_id": record_id}, 

132 ) 

133 return [] 

134 

135 @staticmethod 

136 def get_recent_changes( 

137 session: Session, 

138 limit: int = 100, 

139 ) -> list[AuditLog]: 

140 """Retrieve the most recent audit log entries across all tables. 

141 

142 Args: 

143 session: Database session. 

144 limit: Maximum number of entries to return (default 100). 

145 

146 Returns: 

147 List of recent AuditLog entries, newest first. 

148 """ 

149 try: 

150 return ( 

151 session.query(AuditLog) 

152 .order_by(desc(AuditLog.created_at)) 

153 .limit(limit) 

154 .all() 

155 ) 

156 except Exception: 

157 logger.exception("Failed to query recent audit log entries") 

158 return []