Coverage for /usr/local/lib/python3.12/site-packages/prefect/settings/models/_defaults.py: 44%
79 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1from __future__ import annotations 1a
3from pathlib import Path 1a
4from typing import TYPE_CHECKING, Any 1a
6from pydantic import SecretStr, ValidationInfo 1a
8if TYPE_CHECKING: 8 ↛ 9line 8 didn't jump to line 9 because the condition on line 8 was never true1a
9 from prefect.settings.models.root import Settings
12def default_profiles_path(values: dict[str, Any]) -> Path: 1a
13 """Default profiles_path based on home directory."""
14 home = values.get("home", Path("~/.prefect").expanduser()) 1a
15 if not isinstance(home, Path): 15 ↛ 16line 15 didn't jump to line 16 because the condition on line 15 was never true1a
16 home = Path("~/.prefect").expanduser()
17 return home / "profiles.toml" 1a
20def substitute_home_template(v: Any, info: ValidationInfo) -> Any: 1a
21 """Validator that substitutes $PREFECT_HOME in a path string if present."""
22 home_path = info.data.get("home") 1a
24 path_str: str | None = None 1a
25 if isinstance(v, Path): 25 ↛ 27line 25 didn't jump to line 27 because the condition on line 25 was always true1a
26 return v 1a
27 elif isinstance(v, str):
28 path_str = v
29 elif v is None:
30 return None
31 else:
32 return v
34 if path_str and "$PREFECT_HOME" in path_str:
35 if home_path and isinstance(home_path, Path):
36 resolved_str = path_str.replace("$PREFECT_HOME", str(home_path))
37 try:
38 return Path(resolved_str)
39 except Exception as e:
40 raise ValueError(
41 f"Error creating path after substituting $PREFECT_HOME: {e}"
42 ) from e
43 else:
44 raise ValueError(
45 f'Cannot resolve $PREFECT_HOME in "{path_str}" because '
46 f"PREFECT_HOME setting ({home_path!r}) is not a valid resolved path."
47 )
49 return path_str
52def default_local_storage_path(values: dict[str, Any]) -> Path: 1a
53 """Default local_storage_path based on home directory."""
54 home = values.get("home") 1a
55 if not isinstance(home, Path): 55 ↛ 57line 55 didn't jump to line 57 because the condition on line 55 was always true1a
56 home = Path("~/.prefect").expanduser() 1a
57 return home / "storage" 1a
60def default_memo_store_path(values: dict[str, Any]) -> Path: 1a
61 """Default memo_store_path based on home directory."""
62 home = values.get("home") 1a
63 if not isinstance(home, Path): 63 ↛ 65line 63 didn't jump to line 65 because the condition on line 63 was always true1a
64 home = Path("~/.prefect").expanduser() 1a
65 return home / "memo_store.toml" 1a
68def default_logging_config_path(values: dict[str, Any]) -> Path: 1a
69 """Default logging_config_path based on home directory."""
70 home = values.get("home") 1a
71 if not isinstance(home, Path): 71 ↛ 73line 71 didn't jump to line 73 because the condition on line 71 was always true1a
72 home = Path("~/.prefect").expanduser() 1a
73 return home / "logging.yml" 1a
76def default_database_connection_url(settings: "Settings") -> SecretStr: 1a
77 value: str = f"sqlite+aiosqlite:///{settings.home}/prefect.db" 1a
78 if settings.server.database.driver == "postgresql+asyncpg": 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true1a
79 required = [
80 "host",
81 "user",
82 "name",
83 "password",
84 ]
85 missing = [
86 attr for attr in required if getattr(settings.server.database, attr) is None
87 ]
88 if missing:
89 raise ValueError(
90 f"Missing required database connection settings: {', '.join(missing)}"
91 )
93 from sqlalchemy import URL
95 value = URL(
96 drivername=settings.server.database.driver,
97 host=settings.server.database.host,
98 port=settings.server.database.port or 5432,
99 username=settings.server.database.user,
100 password=(
101 settings.server.database.password.get_secret_value()
102 if settings.server.database.password
103 else None
104 ),
105 database=settings.server.database.name,
106 query=[], # type: ignore
107 ).render_as_string(hide_password=False)
109 elif settings.server.database.driver == "sqlite+aiosqlite": 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true1a
110 if settings.server.database.name:
111 value = (
112 f"{settings.server.database.driver}:///{settings.server.database.name}"
113 )
114 else:
115 value = f"sqlite+aiosqlite:///{settings.home}/prefect.db"
117 elif settings.server.database.driver: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true1a
118 raise ValueError(
119 f"Unsupported database driver: {settings.server.database.driver}"
120 )
121 return SecretStr(value) 1a
124def default_ui_url(settings: "Settings") -> str | None: 1a
125 value = settings.ui_url 1abc
126 if value is not None: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true1abc
127 return value
129 # Otherwise, infer a value from the API URL
130 ui_url = api_url = settings.api.url 1abc
132 if not api_url: 132 ↛ 134line 132 didn't jump to line 134 because the condition on line 132 was always true1abc
133 return None 1abc
134 assert ui_url is not None
136 cloud_url = settings.cloud.api_url
137 cloud_ui_url = settings.cloud.ui_url
138 if api_url.startswith(cloud_url) and cloud_ui_url:
139 ui_url = ui_url.replace(cloud_url, cloud_ui_url)
141 if ui_url.endswith("/api"):
142 # Handles open-source APIs
143 ui_url = ui_url[:-4]
145 # Handles Cloud APIs with content after `/api`
146 ui_url = ui_url.replace("/api/", "/")
148 # Update routing
149 ui_url = ui_url.replace("/accounts/", "/account/")
150 ui_url = ui_url.replace("/workspaces/", "/workspace/")
152 return ui_url