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

1from __future__ import annotations 1 ctx1a

2 

3from pathlib import Path 1 ctx1a

4from typing import TYPE_CHECKING, Any 1 ctx1a

5 

6from pydantic import SecretStr, ValidationInfo 1 ctx1a

7 

8if TYPE_CHECKING: 8 ↛ 9line 8 didn't jump to line 9 because the condition on line 8 was never true1 ctx1a

9 from prefect.settings.models.root import Settings 

10 

11 

12def default_profiles_path(values: dict[str, Any]) -> Path: 1 ctx1a

13 """Default profiles_path based on home directory.""" 

14 home = values.get("home", Path("~/.prefect").expanduser()) 1 ctx1a

15 if not isinstance(home, Path): 15 ↛ 16line 15 didn't jump to line 16 because the condition on line 15 was never true1 ctx1a

16 home = Path("~/.prefect").expanduser() 

17 return home / "profiles.toml" 1 ctx1a

18 

19 

20def substitute_home_template(v: Any, info: ValidationInfo) -> Any: 1 ctx1a

21 """Validator that substitutes $PREFECT_HOME in a path string if present.""" 

22 home_path = info.data.get("home") 1 ctx1a

23 

24 path_str: str | None = None 1 ctx1a

25 if isinstance(v, Path): 25 ↛ 27line 25 didn't jump to line 27 because the condition on line 25 was always true1 ctx1a

26 return v 1 ctx1a

27 elif isinstance(v, str): 

28 path_str = v 

29 elif v is None: 

30 return None 

31 else: 

32 return v 

33 

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 ) 

48 

49 return path_str 

50 

51 

52def default_local_storage_path(values: dict[str, Any]) -> Path: 1 ctx1a

53 """Default local_storage_path based on home directory.""" 

54 home = values.get("home") 1 ctx1a

55 if not isinstance(home, Path): 55 ↛ 57line 55 didn't jump to line 57 because the condition on line 55 was always true1 ctx1a

56 home = Path("~/.prefect").expanduser() 1 ctx1a

57 return home / "storage" 1 ctx1a

58 

59 

60def default_memo_store_path(values: dict[str, Any]) -> Path: 1 ctx1a

61 """Default memo_store_path based on home directory.""" 

62 home = values.get("home") 1 ctx1a

63 if not isinstance(home, Path): 63 ↛ 65line 63 didn't jump to line 65 because the condition on line 63 was always true1 ctx1a

64 home = Path("~/.prefect").expanduser() 1 ctx1a

65 return home / "memo_store.toml" 1 ctx1a

66 

67 

68def default_logging_config_path(values: dict[str, Any]) -> Path: 1 ctx1a

69 """Default logging_config_path based on home directory.""" 

70 home = values.get("home") 1 ctx1a

71 if not isinstance(home, Path): 71 ↛ 73line 71 didn't jump to line 73 because the condition on line 71 was always true1 ctx1a

72 home = Path("~/.prefect").expanduser() 1 ctx1a

73 return home / "logging.yml" 1 ctx1a

74 

75 

76def default_database_connection_url(settings: "Settings") -> SecretStr: 1 ctx1a

77 value: str = f"sqlite+aiosqlite:///{settings.home}/prefect.db" 1 ctx1a

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 true1 ctx1a

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 ) 

92 

93 from sqlalchemy import URL 

94 

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) 

108 

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 true1 ctx1a

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" 

116 

117 elif settings.server.database.driver: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true1 ctx1a

118 raise ValueError( 

119 f"Unsupported database driver: {settings.server.database.driver}" 

120 ) 

121 return SecretStr(value) 1 ctx1a

122 

123 

124def default_ui_url(settings: "Settings") -> str | None: 1 ctx1a

125 value = settings.ui_url 1 ctx1a

126 if value is not None: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true1 ctx1a

127 return value 

128 

129 # Otherwise, infer a value from the API URL 

130 ui_url = api_url = settings.api.url 1 ctx1a

131 

132 if not api_url: 132 ↛ 134line 132 didn't jump to line 134 because the condition on line 132 was always true1 ctx1a

133 return None 1 ctx1a

134 assert ui_url is not None 

135 

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) 

140 

141 if ui_url.endswith("/api"): 

142 # Handles open-source APIs 

143 ui_url = ui_url[:-4] 

144 

145 # Handles Cloud APIs with content after `/api` 

146 ui_url = ui_url.replace("/api/", "/") 

147 

148 # Update routing 

149 ui_url = ui_url.replace("/accounts/", "/account/") 

150 ui_url = ui_url.replace("/workspaces/", "/workspace/") 

151 

152 return ui_url