Coverage for async-durable-execution/src/async_durable_execution/logger.py: 91%

1"""Logging helpers for durable execution contexts.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6 

7from .exceptions import ValidationError 

8from .primitive.base import OperationContext 

9from .context import _current_context 

10 

11 

12class DurableContextFilter(logging.Filter): 

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

14 

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

16 context = _current_context.get() 

17 if context is None: 

18 return True 

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

20 return True 

21 

22 if _is_replaying(context): 

23 return False 

24 

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

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

27 setattr(record, key, value) 

28 return True 

29 

30 

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

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

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

34 execution_arn = context.durable_execution_arn 

35 if execution_arn: 

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

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

38 extra["executionArn"] = execution_arn 

39 parent_id = context.parent_id 

40 if parent_id: 

41 extra["parentId"] = context.parent_id 

42 operation_id = context.operation_id 

43 if operation_id: 

44 extra["operationId"] = context.operation_id 

45 operation_name = context.operation_name 

46 if operation_name: 

47 extra["operationName"] = context.operation_name 

48 

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

50 if callback_id: 

51 extra["callbackId"] = callback_id 

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

53 if attempt is not None: 

54 extra["attempt"] = attempt 

55 return extra 

56 

57 

58def configure_durable_logger(logger): 

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

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

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

62 if not callable(add_filter): 

63 return logger 

64 

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

66 add_filter(DurableContextFilter()) 

67 

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

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

70 handler.addFilter(DurableContextFilter()) 

71 return logger 

72 

73 

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

75 if context.execution_state is None: 

76 raise ValidationError( 

77 "The execution state is None", 

78 ) 

79 return bool(context.is_replaying()) 

80 

81 

82__all__ = [ 

83 "DurableContextFilter", 

84 "build_context_log_extra", 

85 "configure_durable_logger", 

86]