Coverage for /usr/local/lib/python3.12/site-packages/prefect/logging/configuration.py: 62%

54 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 13:38 +0000

1from __future__ import annotations 1a

2 

3import logging 1a

4import logging.config 1a

5import os 1a

6import re 1a

7import string 1a

8import warnings 1a

9from functools import partial 1a

10from pathlib import Path 1a

11from typing import Any, Callable 1a

12 

13import yaml 1a

14 

15from prefect.settings import ( 1a

16 PREFECT_LOGGING_EXTRA_LOGGERS, 

17 PREFECT_LOGGING_SETTINGS_PATH, 

18 get_current_settings, 

19) 

20from prefect.utilities.collections import dict_to_flatdict, flatdict_to_dict 1a

21 

22# This path will be used if `PREFECT_LOGGING_SETTINGS_PATH` is null 

23DEFAULT_LOGGING_SETTINGS_PATH = Path(__file__).parent / "logging.yml" 1a

24 

25# Stores the configuration used to setup logging in this Python process 

26PROCESS_LOGGING_CONFIG: dict[str, Any] = {} 1a

27 

28# Regex call to replace non-alphanumeric characters to '_' to create a valid env var 

29to_envvar: Callable[[str], str] = partial(re.sub, re.compile(r"[^0-9a-zA-Z]+"), "_") 1a

30 

31 

32def load_logging_config(path: Path) -> dict[str, Any]: 1a

33 """ 

34 Loads logging configuration from a path allowing override from the environment 

35 """ 

36 current_settings = get_current_settings() 1a

37 template = string.Template(path.read_text()) 1a

38 

39 with warnings.catch_warnings(): 1a

40 warnings.filterwarnings("ignore", category=DeprecationWarning) 1a

41 config = yaml.safe_load( 1a

42 # Substitute settings into the template in format $SETTING / ${SETTING} 

43 template.substitute( 

44 current_settings.to_environment_variables(include_aliases=True) 

45 ) 

46 ) 

47 

48 # Load overrides from the environment 

49 flat_config = dict_to_flatdict(config) 1a

50 

51 for key_tup, val in flat_config.items(): 1a

52 env_val = os.environ.get( 1a

53 # Generate a valid environment variable with nesting indicated with '_' 

54 to_envvar("PREFECT_LOGGING_" + "_".join(key_tup)).upper() 

55 ) 

56 if env_val: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true1a

57 if isinstance(val, list): 

58 val = env_val.split(",") 

59 else: 

60 val = env_val 

61 

62 # reassign the updated value 

63 flat_config[key_tup] = val 1a

64 

65 return flatdict_to_dict(flat_config) 1a

66 

67 

68def setup_logging(incremental: bool | None = None) -> dict[str, Any]: 1a

69 """ 

70 Sets up logging. 

71 

72 Returns the config used. 

73 """ 

74 global PROCESS_LOGGING_CONFIG 

75 

76 # If the user has specified a logging path and it exists we will ignore the 

77 # default entirely rather than dealing with complex merging 

78 config = load_logging_config( 1a

79 ( 

80 PREFECT_LOGGING_SETTINGS_PATH.value() 

81 if PREFECT_LOGGING_SETTINGS_PATH.value().exists() 

82 else DEFAULT_LOGGING_SETTINGS_PATH 

83 ) 

84 ) 

85 

86 incremental = ( 1a

87 incremental if incremental is not None else bool(PROCESS_LOGGING_CONFIG) 

88 ) 

89 

90 # Perform an incremental update if setup has already been run 

91 config.setdefault("incremental", incremental) 1a

92 

93 root_logger = logging.getLogger() 1a

94 if root_logger.handlers and not incremental: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true1a

95 from prefect.logging.handlers import PrefectConsoleHandler 

96 

97 has_user_handlers = any( 

98 hasattr(handler, "formatter") 

99 and handler.formatter is not None 

100 and not isinstance(handler, PrefectConsoleHandler) 

101 and not handler.__class__.__name__.startswith("_") 

102 and not handler.__class__.__name__ == "LogCaptureHandler" 

103 for handler in root_logger.handlers 

104 ) 

105 if has_user_handlers: 

106 config.pop("root", None) 

107 

108 try: 1a

109 logging.config.dictConfig(config) 1a

110 except ValueError: 

111 if incremental: 

112 setup_logging(incremental=False) 

113 

114 # Copy configuration of the 'prefect.extra' logger to the extra loggers 

115 extra_config = logging.getLogger("prefect.extra") 1a

116 

117 for logger_name in PREFECT_LOGGING_EXTRA_LOGGERS.value(): 117 ↛ 118line 117 didn't jump to line 118 because the loop on line 117 never started1a

118 logger = logging.getLogger(logger_name) 

119 if not config["incremental"]: 

120 for handler in extra_config.handlers: 

121 logger.addHandler(handler) 

122 

123 PROCESS_LOGGING_CONFIG.update(config) 1a

124 

125 return config 1a