Coverage for src/lexigram/admin/services/activity_logger.py: 47%

1"""Activity logging service for admin user actions.""" 

2 

3from __future__ import annotations 

4 

5import contextvars 

6from datetime import UTC, datetime 

7from typing import Any 

8 

9from lexigram.di.decorators import inject 

10from lexigram.logging import get_logger 

11 

12# Request-scoped in-memory dict, isolated per async context. 

13# Populated externally (e.g. by admin middleware) when user context is available. 

14_request_cache_var: contextvars.ContextVar[dict[str, Any] | None] = ( 

15 contextvars.ContextVar( 

16 "admin_request_cache", 

17 default=None, 

18 ) 

19) 

20 

21 

22def get_request_cache() -> dict[str, Any]: 

23 """Return the request-scoped cache dict for the current async context.""" 

24 cache = _request_cache_var.get() 

25 return cache if cache is not None else {} 

26 

27 

28logger = get_logger(__name__) 

29 

30 

31@inject 

32class ActivityLogger: 

33 """Service for tracking and retrieving admin user activity.""" 

34 

35 def __init__(self, capacity: int = 1000): 

36 self._capacity = capacity 

37 self._activities: list[dict[str, Any]] = [] 

38 

39 def log( 

40 self, 

41 action: str, 

42 resource: str, 

43 resource_id: str | None = None, 

44 details: dict[str, Any] | None = None, 

45 success: bool = True, 

46 error: str | None = None, 

47 ) -> None: 

48 """Log an admin activity. 

49 

50 This captures context from the current request if available. 

51 """ 

52 request_cache = get_request_cache() 

53 user_id = request_cache.get("user_id", "system") 

54 

55 entry = { 

56 "timestamp": datetime.now(UTC).isoformat(), 

57 "user_id": user_id, 

58 "action": action, 

59 "resource": resource, 

60 "resource_id": resource_id, 

61 "details": details or {}, 

62 "success": success, 

63 "error": error, 

64 } 

65 

66 # Append to in-memory list (last N activities) 

67 self._activities.append(entry) 

68 if len(self._activities) > self._capacity: 

69 self._activities.pop(0) 

70 

71 # Also log to standard logger 

72 log_msg = f"Admin Action: {action} on {resource}/{resource_id} by {user_id} - {'Success' if success else 'Failed'}" 

73 if error: 

74 log_msg += f": {error}" 

75 logger.info(log_msg) 

76 

77 def get_recent(self, limit: int = 50) -> list[dict[str, Any]]: 

78 """Get recent activities.""" 

79 return self._activities[-limit:] 

80 

81 def clear(self) -> None: 

82 """Clear the activity history.""" 

83 self._activities.clear() 

84 

85 

86_instance: ActivityLogger | None = None 

87 

88 

89def get_activity_logger() -> ActivityLogger: 

90 """Get global activity logger singleton.""" 

91 global _instance 

92 if _instance is None: 

93 _instance = ActivityLogger() 

94 return _instance