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

54 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 10:48 +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() 1acdb

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

38 

39 with warnings.catch_warnings(): 1acdb

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

41 config = yaml.safe_load( 1acdb

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) 1acdb

50 

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

52 env_val = os.environ.get( 1acdb

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 true1acdb

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 1acdb

64 

65 return flatdict_to_dict(flat_config) 1acdb

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( 1acdb

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 = ( 1acdb

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) 1acdb

92 

93 root_logger = logging.getLogger() 1acdb

94 if root_logger.handlers and not incremental: 1acdb

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: 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true

106 config.pop("root", None) 

107 

108 try: 1acdb

109 logging.config.dictConfig(config) 1acdb

110 except ValueError: 

111 if incremental: 111 ↛ 115line 111 didn't jump to line 115 because the condition on line 111 was always true

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") 1acdb

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 started1acdb

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) 1acdb

124 

125 return config 1acdb