Coverage for /usr/local/lib/python3.12/site-packages/prefect/settings/legacy.py: 78%

92 statements  

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

1import inspect 1a

2import os 1a

3from functools import cache 1a

4from typing import Any, Dict, Optional, Set, Type, get_args 1a

5 

6from pydantic import AliasChoices 1a

7from pydantic_settings import BaseSettings 1a

8from typing_extensions import Self 1a

9 

10from prefect.settings.base import PrefectBaseSettings 1a

11from prefect.settings.constants import _SECRET_TYPES # type: ignore[reportPrivateUsage] 1a

12from prefect.settings.context import get_current_settings 1a

13from prefect.settings.models.root import Settings 1a

14 

15 

16class Setting: 1a

17 """Mimics the old Setting object for compatibility with existing code.""" 

18 

19 def __init__( 1a

20 self, name: str, default: Any, type_: Any, accessor: Optional[str] = None 

21 ): 

22 self._name = name 1a

23 self._default = default 1a

24 self._type = type_ 1a

25 if accessor is None: 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was never true1a

26 self.accessor: str = _env_var_to_accessor(name) 

27 else: 

28 self.accessor: str = accessor 1a

29 

30 @property 1a

31 def name(self) -> str: 1a

32 return self._name 1albijcmdekfgh

33 

34 @property 1a

35 def is_secret(self) -> bool: 1a

36 if self._type in _SECRET_TYPES: 

37 return True 

38 for secret_type in _SECRET_TYPES: 

39 if secret_type in get_args(self._type): 

40 return True 

41 return False 

42 

43 def default(self) -> Any: 1a

44 return self._default 

45 

46 def value(self: Self) -> Any: 1a

47 if ( 47 ↛ 51line 47 didn't jump to line 51 because the condition on line 47 was never true

48 self.name == "PREFECT_TEST_SETTING" 

49 or self.name == "PREFECT_TESTING_TEST_SETTING" 

50 ): 

51 if ( 

52 "PREFECT_TEST_MODE" in os.environ 

53 or "PREFECT_TESTING_TEST_MODE" in os.environ 

54 ): 

55 return get_current_settings().testing.test_setting 

56 else: 

57 return None 

58 

59 return self.value_from(get_current_settings()) 1albijcmdekfgh

60 

61 def value_from(self: Self, settings: Settings) -> Any: 1a

62 path = self.accessor.split(".") 1albijcmdekfgh

63 current_value = settings 1albijcmdekfgh

64 for key in path: 1albijcmdekfgh

65 current_value = getattr(current_value, key, None) 1albijcmdekfgh

66 if isinstance(current_value, _SECRET_TYPES): 1albijcmdekfgh

67 return current_value.get_secret_value() # type: ignore 1albijcdekfgh

68 return current_value 1abijcmdekfgh

69 

70 def __bool__(self) -> bool: 1a

71 return bool(self.value()) 1abcdefgh

72 

73 def __str__(self) -> str: 1a

74 return str(self.value()) 

75 

76 def __repr__(self) -> str: 1a

77 return f"<{self.name}: {self._type!r}>" 

78 

79 def __eq__(self, __o: object) -> bool: 1a

80 return __o.__eq__(self.value()) 

81 

82 def __hash__(self) -> int: 1a

83 return hash((type(self), self.name)) 1a

84 

85 

86def _env_var_to_accessor(env_var: str) -> str: 1a

87 """ 

88 Convert an environment variable name to a settings accessor. 

89 """ 

90 if (field := _get_settings_fields(Settings).get(env_var)) is not None: 

91 return field.accessor 

92 return env_var.replace("PREFECT_", "").lower() 

93 

94 

95@cache 1a

96def _get_valid_setting_names(cls: type[BaseSettings]) -> Set[str]: 1a

97 """ 

98 A set of valid setting names, e.g. "PREFECT_API_URL" or "PREFECT_API_KEY". 

99 """ 

100 settings_fields: set[str] = set() 1a

101 for field_name, field in cls.model_fields.items(): 1a

102 if inspect.isclass(field.annotation) and issubclass( 1a

103 field.annotation, PrefectBaseSettings 

104 ): 

105 settings_fields.update(_get_valid_setting_names(field.annotation)) 1a

106 else: 

107 if field.validation_alias and isinstance( 1a

108 field.validation_alias, AliasChoices 

109 ): 

110 for alias in field.validation_alias.choices: 1a

111 if not isinstance(alias, str): 1a

112 continue 1a

113 settings_fields.add(alias.upper()) 1a

114 else: 

115 settings_fields.add( 1a

116 f"{cls.model_config.get('env_prefix')}{field_name.upper()}" 

117 ) 

118 return settings_fields 1a

119 

120 

121@cache 1a

122def _get_settings_fields( 1a

123 settings: Type[BaseSettings], accessor_prefix: Optional[str] = None 

124) -> Dict[str, "Setting"]: 

125 """Get the settings fields for the settings object""" 

126 settings_fields: dict[str, Setting] = {} 1a

127 for field_name, field in settings.model_fields.items(): 1a

128 if inspect.isclass(field.annotation) and issubclass( 1a

129 field.annotation, PrefectBaseSettings 

130 ): 

131 accessor = ( 1a

132 field_name 

133 if accessor_prefix is None 

134 else f"{accessor_prefix}.{field_name}" 

135 ) 

136 settings_fields.update(_get_settings_fields(field.annotation, accessor)) 1a

137 else: 

138 accessor = ( 1a

139 field_name 

140 if accessor_prefix is None 

141 else f"{accessor_prefix}.{field_name}" 

142 ) 

143 if field.validation_alias and isinstance( 1a

144 field.validation_alias, AliasChoices 

145 ): 

146 for alias in field.validation_alias.choices: 1a

147 if not isinstance(alias, str): 1a

148 continue 1a

149 setting = Setting( 1a

150 name=alias.upper(), 

151 default=field.default, 

152 type_=field.annotation, 

153 accessor=accessor, 

154 ) 

155 settings_fields[setting.name] = setting 1a

156 settings_fields[setting.accessor] = setting 1a

157 else: 

158 setting = Setting( 1a

159 name=f"{settings.model_config.get('env_prefix')}{field_name.upper()}", 

160 default=field.default, 

161 type_=field.annotation, 

162 accessor=accessor, 

163 ) 

164 settings_fields[setting.name] = setting 1a

165 settings_fields[setting.accessor] = setting 1a

166 

167 return settings_fields 1a