Coverage for async_durable_execution/_core/logger.py: 99%

1"""Core logging helpers for durable execution contexts.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6from typing import TypeVar 

7 

8from .context import OperationContext, _current_context 

9from .exceptions import ValidationError 

10 

11_LoggerT = TypeVar("_LoggerT") 

12 

13 

14class DurableContextFilter(logging.Filter): 

15 """Add durable execution metadata from the active contextvar to log records.""" 

16 

17 def filter(self, record: logging.LogRecord) -> bool: 

18 context = _current_context.get() 

19 if context is None: 

20 return True 

21 if not hasattr(context, "execution_state"): 

22 return True 

23 

24 if _is_replaying(context): 

25 return False 

26 

27 for key, value in build_context_log_extra(context).items(): 

28 if value is not None and not hasattr(record, key): 

29 setattr(record, key, value) 

30 return True 

31 

32 

33def build_context_log_extra(context: OperationContext) -> dict[str, object]: 

34 """Build structured log fields from the active execution context.""" 

35 extra: dict[str, object] = {} 

36 execution_arn = context.durable_execution_arn 

37 if execution_arn: 37 ↛ 41line 37 didn't jump to line 41 because the condition on line 37 was always true

38 # `executionArn` is used here while `durableExecutionArn` is used everywhere else because 

39 # that's what the Lambda Console expects in log records. 

40 extra["executionArn"] = execution_arn 

41 parent_id = context.parent_id 

42 if parent_id: 

43 extra["parentId"] = context.parent_id 

44 operation_id = context.operation_id 

45 if operation_id: 

46 extra["operationId"] = context.operation_id 

47 operation_name = context.operation_name 

48 if operation_name: 

49 extra["operationName"] = context.operation_name 

50 

51 callback_id = getattr(context, "callback_id", None) 

52 if callback_id: 

53 extra["callbackId"] = callback_id 

54 attempt = getattr(context, "attempt", None) 

55 if attempt is not None: 

56 extra["attempt"] = attempt 

57 return extra 

58 

59 

60def configure_durable_logger(logger: _LoggerT) -> _LoggerT: 

61 """Attach DurableContextFilter to a stdlib-compatible logger and handlers.""" 

62 add_filter = getattr(logger, "addFilter", None) 

63 filters = getattr(logger, "filters", ()) 

64 if not callable(add_filter): 

65 return logger 

66 

67 if not any(isinstance(item, DurableContextFilter) for item in filters): 

68 add_filter(DurableContextFilter()) 

69 

70 for handler in getattr(logger, "handlers", ()): 

71 if not any(isinstance(item, DurableContextFilter) for item in handler.filters): 

72 handler.addFilter(DurableContextFilter()) 

73 return logger 

74 

75 

76def _is_replaying(context: OperationContext) -> bool: 

77 if context.execution_state is None: 

78 raise ValidationError( 

79 "The execution state is None", 

80 ) 

81 return bool(context.is_replaying()) 

82 

83 

84__all__ = [ 

85 "DurableContextFilter", 

86 "build_context_log_extra", 

87 "configure_durable_logger", 

88]