Coverage for /usr/local/lib/python3.12/site-packages/prefect/settings/models/root.py: 58%

150 statements  

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

1import warnings 1a

2from pathlib import Path 1a

3from typing import ( 1a

4 TYPE_CHECKING, 

5 Annotated, 

6 Any, 

7 ClassVar, 

8 Iterable, 

9 Mapping, 

10 Optional, 

11) 

12from urllib.parse import urlparse 1a

13 

14from pydantic import BeforeValidator, Field, SecretStr, model_validator 1a

15from pydantic_settings import SettingsConfigDict 1a

16from typing_extensions import Self 1a

17 

18from prefect.settings.base import PrefectBaseSettings, build_settings_config 1a

19from prefect.settings.models.tasks import TasksSettings 1a

20from prefect.settings.models.testing import TestingSettings 1a

21from prefect.settings.models.worker import WorkerSettings 1a

22from prefect.utilities.collections import deep_merge_dicts, set_in_dict 1a

23 

24from ._defaults import ( 1a

25 default_database_connection_url, 

26 default_profiles_path, 

27 default_ui_url, 

28 substitute_home_template, 

29) 

30from .api import APISettings 1a

31from .cli import CLISettings 1a

32from .client import ClientSettings 1a

33from .cloud import CloudSettings 1a

34from .deployments import DeploymentsSettings 1a

35from .experiments import ExperimentsSettings 1a

36from .flows import FlowsSettings 1a

37from .internal import InternalSettings 1a

38from .logging import LoggingSettings 1a

39from .results import ResultsSettings 1a

40from .runner import RunnerSettings 1a

41from .server import ServerSettings 1a

42 

43if TYPE_CHECKING: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true1a

44 from prefect.settings.legacy import Setting 

45 

46 

47class Settings(PrefectBaseSettings): 1a

48 """ 

49 Settings for Prefect using Pydantic settings. 

50 

51 See https://docs.pydantic.dev/latest/concepts/pydantic_settings 

52 """ 

53 

54 model_config: ClassVar[SettingsConfigDict] = build_settings_config() 1a

55 

56 home: Annotated[Path, BeforeValidator(lambda x: Path(x).expanduser())] = Field( 1a

57 default=Path("~") / ".prefect", 

58 description="The path to the Prefect home directory. Defaults to ~/.prefect", 

59 ) 

60 

61 profiles_path: Annotated[Path, BeforeValidator(substitute_home_template)] = Field( 1a

62 default_factory=default_profiles_path, 

63 # Escaped backslashes are to prevent latex rendering 

64 description=( 

65 "The path to a profiles configuration file. Supports \\$PREFECT_HOME templating." 

66 " Defaults to \\$PREFECT_HOME/profiles.toml." 

67 ), 

68 ) 

69 

70 debug_mode: bool = Field( 1a

71 default=False, 

72 description="If True, enables debug mode which may provide additional logging and debugging features.", 

73 ) 

74 

75 api: APISettings = Field( 1a

76 default_factory=APISettings, 

77 description="Settings for interacting with the Prefect API", 

78 ) 

79 

80 cli: CLISettings = Field( 1a

81 default_factory=CLISettings, 

82 description="Settings for controlling CLI behavior", 

83 ) 

84 

85 client: ClientSettings = Field( 1a

86 default_factory=ClientSettings, 

87 description="Settings for controlling API client behavior", 

88 ) 

89 

90 cloud: CloudSettings = Field( 1a

91 default_factory=CloudSettings, 

92 description="Settings for interacting with Prefect Cloud", 

93 ) 

94 

95 deployments: DeploymentsSettings = Field( 1a

96 default_factory=DeploymentsSettings, 

97 description="Settings for configuring deployments defaults", 

98 ) 

99 

100 experiments: ExperimentsSettings = Field( 1a

101 default_factory=ExperimentsSettings, 

102 description="Settings for controlling experimental features", 

103 ) 

104 

105 flows: FlowsSettings = Field( 1a

106 default_factory=FlowsSettings, 

107 description="Settings for controlling flow behavior", 

108 ) 

109 

110 internal: InternalSettings = Field( 1a

111 default_factory=InternalSettings, 

112 description="Settings for internal Prefect machinery", 

113 ) 

114 

115 logging: LoggingSettings = Field( 1a

116 default_factory=LoggingSettings, 

117 description="Settings for controlling logging behavior", 

118 ) 

119 

120 results: ResultsSettings = Field( 1a

121 default_factory=ResultsSettings, 

122 description="Settings for controlling result storage behavior", 

123 ) 

124 

125 runner: RunnerSettings = Field( 1a

126 default_factory=RunnerSettings, 

127 description="Settings for controlling runner behavior", 

128 ) 

129 

130 server: ServerSettings = Field( 1a

131 default_factory=ServerSettings, 

132 description="Settings for controlling server behavior", 

133 ) 

134 

135 tasks: TasksSettings = Field( 1a

136 default_factory=TasksSettings, 

137 description="Settings for controlling task behavior", 

138 ) 

139 

140 testing: TestingSettings = Field( 1a

141 default_factory=TestingSettings, 

142 description="Settings used during testing", 

143 ) 

144 

145 worker: WorkerSettings = Field( 1a

146 default_factory=WorkerSettings, 

147 description="Settings for controlling worker behavior", 

148 ) 

149 

150 ui_url: Optional[str] = Field( 1a

151 default=None, 

152 description="The URL of the Prefect UI. If not set, the client will attempt to infer it.", 

153 ) 

154 

155 silence_api_url_misconfiguration: bool = Field( 1a

156 default=False, 

157 description=""" 

158 If `True`, disable the warning when a user accidentally misconfigure its `PREFECT_API_URL` 

159 Sometimes when a user manually set `PREFECT_API_URL` to a custom url,reverse-proxy for example, 

160 we would like to silence this warning so we will set it to `FALSE`. 

161 """, 

162 ) 

163 

164 ########################################################################### 

165 # allow deprecated access to PREFECT_SOME_SETTING_NAME 

166 

167 def __getattribute__(self, name: str) -> Any: 1a

168 from prefect.settings.legacy import _env_var_to_accessor 1abcde

169 

170 if name.startswith("PREFECT_"): 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true1abcde

171 accessor = _env_var_to_accessor(name) 

172 warnings.warn( 

173 f"Accessing `Settings().{name}` is deprecated. Use `Settings().{accessor}` instead.", 

174 DeprecationWarning, 

175 stacklevel=2, 

176 ) 

177 path = accessor.split(".") 

178 value = super().__getattribute__(path[0]) 

179 for key in path[1:]: 

180 value = getattr(value, key) 

181 return value 

182 return super().__getattribute__(name) 1abcde

183 

184 ########################################################################### 

185 

186 @model_validator(mode="after") 1a

187 def post_hoc_settings(self) -> Self: 1a

188 """Handle remaining complex default assignments that aren't yet migrated to dependent settings. 

189 

190 With Pydantic 2.10's dependent settings feature, we've migrated simple path-based defaults 

191 to use default_factory. The remaining items here require access to the full Settings instance 

192 or have complex interdependencies that will be migrated in future PRs. 

193 """ 

194 if self.ui_url is None: 194 ↛ 197line 194 didn't jump to line 197 because the condition on line 194 was always true1a

195 self.ui_url = default_ui_url(self) 1a

196 self.__pydantic_fields_set__.remove("ui_url") 1a

197 if self.server.ui.api_url is None: 1a

198 if self.api.url: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true1a

199 self.server.ui.api_url = self.api.url 

200 self.server.ui.__pydantic_fields_set__.remove("api_url") 

201 else: 

202 self.server.ui.api_url = ( 1a

203 f"http://{self.server.api.host}:{self.server.api.port}/api" 

204 ) 

205 self.server.ui.__pydantic_fields_set__.remove("api_url") 1a

206 if self.debug_mode or self.testing.test_mode: 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true1a

207 self.logging.level = "DEBUG" 

208 self.internal.logging_level = "DEBUG" 

209 self.logging.__pydantic_fields_set__.remove("level") 

210 self.internal.__pydantic_fields_set__.remove("logging_level") 

211 

212 # Set default database connection URL if not provided 

213 if self.server.database.connection_url is None: 1a

214 self.server.database.connection_url = default_database_connection_url(self) 1a

215 self.server.database.__pydantic_fields_set__.remove("connection_url") 1a

216 db_url = self.server.database.connection_url.get_secret_value() 1a

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

218 "PREFECT_API_DATABASE_PASSWORD" in db_url 

219 or "PREFECT_SERVER_DATABASE_PASSWORD" in db_url 

220 ): 

221 if self.server.database.password is None: 

222 raise ValueError( 

223 "database password is None - please set PREFECT_SERVER_DATABASE_PASSWORD" 

224 ) 

225 db_url = db_url.replace( 

226 "${PREFECT_API_DATABASE_PASSWORD}", 

227 self.server.database.password.get_secret_value() 

228 if self.server.database.password 

229 else "", 

230 ) 

231 db_url = db_url.replace( 

232 "${PREFECT_SERVER_DATABASE_PASSWORD}", 

233 self.server.database.password.get_secret_value() 

234 if self.server.database.password 

235 else "", 

236 ) 

237 self.server.database.connection_url = SecretStr(db_url) 

238 self.server.database.__pydantic_fields_set__.remove("connection_url") 

239 

240 if self.connected_to_cloud: 240 ↛ 242line 240 didn't jump to line 242 because the condition on line 240 was never true1a

241 # Ensure we don't exceed Cloud's maximum log size 

242 self.logging.to_api.max_log_size = min( 

243 self.logging.to_api.max_log_size, self.cloud.max_log_size 

244 ) 

245 

246 return self 1a

247 

248 @model_validator(mode="after") 1a

249 def emit_warnings(self) -> Self: 1a

250 """More post-hoc validation of settings, including warnings for misconfigurations.""" 

251 if not self.silence_api_url_misconfiguration: 251 ↛ 253line 251 didn't jump to line 253 because the condition on line 251 was always true1a

252 _warn_on_misconfigured_api_url(self) 1a

253 return self 1a

254 

255 @property 1a

256 def connected_to_cloud(self) -> bool: 1a

257 """True when the API URL points at the configured Prefect Cloud API.""" 

258 return bool( 1a

259 self.api.url 

260 and self.cloud.api_url 

261 and self.api.url.startswith(self.cloud.api_url) 

262 ) 

263 

264 ########################################################################## 

265 # Settings methods 

266 

267 def copy_with_update( 1a

268 self: Self, 

269 updates: Optional[Mapping["Setting", Any]] = None, 

270 set_defaults: Optional[Mapping["Setting", Any]] = None, 

271 restore_defaults: Optional[Iterable["Setting"]] = None, 

272 ) -> Self: 

273 """ 

274 Create a new Settings object with validation. 

275 

276 Arguments: 

277 updates: A mapping of settings to new values. Existing values for the 

278 given settings will be overridden. 

279 set_defaults: A mapping of settings to new default values. Existing values for 

280 the given settings will only be overridden if they were not set. 

281 restore_defaults: An iterable of settings to restore to their default values. 

282 

283 Returns: 

284 A new Settings object. 

285 """ 

286 # To restore defaults, we need to resolve the setting path and then 

287 # set the default value on the new settings object. When restoring 

288 # defaults, all settings sources will be ignored. 

289 restore_defaults_obj: dict[str, Any] = {} 1a

290 for r in restore_defaults or []: 290 ↛ 291line 290 didn't jump to line 291 because the loop on line 290 never started1a

291 path = r.accessor.split(".") 

292 model = self 

293 model_cls = model.__class__ 

294 model_fields = model_cls.model_fields 

295 for key in path[:-1]: 

296 model_field = model_fields[key] 

297 model_cls = model_field.annotation 

298 if model_cls is None: 

299 raise ValueError(f"Invalid setting path: {r.accessor}") 

300 model_fields = model_cls.model_fields 

301 

302 model_field = model_fields[path[-1]] 

303 assert model_field is not None, f"Invalid setting path: {r.accessor}" 

304 if hasattr(model_field, "default"): 

305 default = model_field.default 

306 elif ( 

307 hasattr(model_field, "default_factory") and model_field.default_factory 

308 ): 

309 default = model_field.default_factory() 

310 else: 

311 raise ValueError(f"No default value for setting: {r.accessor}") 

312 set_in_dict( 

313 restore_defaults_obj, 

314 r.accessor, 

315 default, 

316 ) 

317 updates = updates or {} 1a

318 set_defaults = set_defaults or {} 1a

319 

320 set_defaults_obj: dict[str, Any] = {} 1a

321 for setting, value in set_defaults.items(): 321 ↛ 322line 321 didn't jump to line 322 because the loop on line 321 never started1a

322 set_in_dict(set_defaults_obj, setting.accessor, value) 

323 

324 updates_obj: dict[str, Any] = {} 1a

325 for setting, value in updates.items(): 1a

326 set_in_dict(updates_obj, setting.accessor, value) 1a

327 

328 new_settings = self.__class__.model_validate( 1a

329 deep_merge_dicts( 

330 set_defaults_obj, 

331 self.model_dump(exclude_unset=True), 

332 restore_defaults_obj, 

333 updates_obj, 

334 ) 

335 ) 

336 return new_settings 1a

337 

338 def hash_key(self) -> str: 1a

339 """ 

340 Return a hash key for the settings object. This is needed since some 

341 settings may be unhashable, like lists. 

342 """ 

343 env_variables = self.to_environment_variables() 1a

344 return str(hash(tuple((key, value) for key, value in env_variables.items()))) 1a

345 

346 

347def _warn_on_misconfigured_api_url(settings: "Settings"): 1a

348 """ 

349 Validator for settings warning if the API URL is misconfigured. 

350 """ 

351 api_url = settings.api.url 1a

352 if api_url is not None: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true1a

353 misconfigured_mappings = { 

354 "app.prefect.cloud": ( 

355 "`PREFECT_API_URL` points to `app.prefect.cloud`. Did you" 

356 " mean `api.prefect.cloud`?" 

357 ), 

358 "account/": ( 

359 "`PREFECT_API_URL` uses `/account/` but should use `/accounts/`." 

360 ), 

361 "workspace/": ( 

362 "`PREFECT_API_URL` uses `/workspace/` but should use `/workspaces/`." 

363 ), 

364 } 

365 warnings_list: list[str] = [] 

366 

367 for misconfig, warning in misconfigured_mappings.items(): 

368 if misconfig in api_url: 

369 warnings_list.append(warning) 

370 

371 parsed_url = urlparse(api_url) 

372 if ( 

373 parsed_url.path 

374 and "api.prefect.cloud" in api_url 

375 and not parsed_url.path.startswith("/api") 

376 ): 

377 warnings_list.append( 

378 "`PREFECT_API_URL` should have `/api` after the base URL." 

379 ) 

380 

381 if warnings_list: 

382 example = 'e.g. PREFECT_API_URL="https://api.prefect.cloud/api/accounts/[ACCOUNT-ID]/workspaces/[WORKSPACE-ID]"' 

383 warnings_list.append(example) 

384 

385 warnings.warn("\n".join(warnings_list), stacklevel=2) 

386 

387 return settings 1a

388 

389 

390def canonical_environment_prefix(settings: "Settings") -> str: 1a

391 return settings.model_config.get("env_prefix") or ""