Coverage for src/fastapi_cloud_cli/commands/apps/link.py: 100%

1import logging 7 ctx1CDEFGHI

2from pathlib import Path 7 ctx1CDEFGHI

3from typing import Annotated, Any, NoReturn 7 ctx1CDEFGHI

4 

5import typer 7 ctx1CDEFGHI

6from pydantic import BaseModel, Field 7 ctx1CDEFGHI

7from rich_toolkit import RichToolkit 7 ctx1CDEFGHI

8from rich_toolkit.menu import Option 7 ctx1CDEFGHI

9 

10from fastapi_cloud_cli._app import cloud_app 7 ctx1CDEFGHI

11from fastapi_cloud_cli.api import APIClient 7 ctx1CDEFGHI

12from fastapi_cloud_cli.commands.apps._app import apps_app 7 ctx1CDEFGHI

13from fastapi_cloud_cli.commands.apps.list import _get_app 7 ctx1CDEFGHI

14from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config, write_app_config 7 ctx1CDEFGHI

15from fastapi_cloud_cli.utils.auth import Identity 7 ctx1CDEFGHI

16from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit, get_rich_toolkit 7 ctx1CDEFGHI

17from fastapi_cloud_cli.utils.execution import JsonOutputOption 7 ctx1CDEFGHI

18 

19logger = logging.getLogger(__name__) 7 ctx1CDEFGHI

20 

21 

22class LinkOutput(BaseModel): 7 ctx1CDEFGHI

23 app_id: str 7 ctx1CDEFGHI

24 team_id: str 7 ctx1CDEFGHI

25 path: Path 7 ctx1CDEFGHI

26 app_name: Annotated[str, Field(exclude=True)] 7 ctx1CDEFGHI

27 config_path: Annotated[Path, Field(exclude=True)] 7 ctx1CDEFGHI

28 

29 

30def _render_link_output(data: LinkOutput, toolkit: RichToolkit) -> None: 10 ctx1CD4EF5GHI6

31 toolkit.print( 7 ctx174859!6

32 f"Linked [bold]{data.path}[/bold] to [bold]{data.app_name}[/bold]", 

33 bullet=False, 

34 ) 

35 toolkit.print(f"Config: [bold]{data.config_path}[/bold]", bullet=False) 7 ctx174859!6

36 

37 

38def _fail_not_logged_in_interactive(toolkit: FastAPIRichToolkit) -> NoReturn: 7 ctx1CDEFGHI

39 toolkit.fail( 7 ctx2EbFbGbHbIbJbKb

40 "not_logged_in", 

41 "You need to be logged in to link an app.", 

42 hint="Run [bold]fastapi cloud login[/] to authenticate.", 

43 ) 

44 

45 

46def _fail_already_linked_interactive(toolkit: FastAPIRichToolkit) -> NoReturn: 7 ctx1CDEFGHI

47 toolkit.fail( 7 ctx2jbkblbmbnbobpb

48 "already_linked", 

49 "This directory is already linked to an app.", 

50 hint="Run [bold]fastapi cloud unlink[/] first to remove the existing configuration.", 

51 ) 

52 

53 

54def _link_app_by_id( 7 ctx1CDEFGHI

55 toolkit: FastAPIRichToolkit, 

56 *, 

57 app_id: str, 

58 path_to_link: Path, 

59 force: bool, 

60) -> None: 

61 if get_app_config(path_to_link) and not force: 42 ctx2+ qb, - 7 . / rb: ; 4 = ? sb@ [ 8 ] ^ tb_ ` 5 { | ub} ~ 9 abbbvbcbdb! ebfbwbgbhb6 ib

62 toolkit.fail( 7 ctx2qbrbsbtbubvbwb

63 "already_linked", 

64 "This directory is already linked to an app.", 

65 hint="Pass --force to replace the existing configuration.", 

66 ) 

67 

68 with APIClient() as client: 35 ctx2+ , - 7 . / : ; 4 = ? @ [ 8 ] ^ _ ` 5 { | } ~ 9 abbbcbdb! ebfbgbhb6 ib

69 with toolkit.progress( 35 ctx2+ , - 7 . / : ; 4 = ? @ [ 8 ] ^ _ ` 5 { | } ~ 9 abbbcbdb! ebfbgbhb6 ib

70 title="Fetching app", 

71 transient=True, 

72 ) as progress: 

73 with client.handle_http_errors( 35 ctx2+ , - 7 . / : ; 4 = ? @ [ 8 ] ^ _ ` 5 { | } ~ 9 abbbcbdb! ebfbgbhb6 ib

74 progress, 

75 default_message="Error fetching app. Please try again later.", 

76 not_found_message="App not found.", 

77 toolkit=toolkit, 

78 ): 

79 app = _get_app(client, app_id) 35 ctx2+ , - 7 . / : ; 4 = ? @ [ 8 ] ^ _ ` 5 { | } ~ 9 abbbcbdb! ebfbgbhb6 ib

80 

81 write_app_config( 35 ctx2+ , - 7 . / : ; 4 = ? @ [ 8 ] ^ _ ` 5 { | } ~ 9 abbbcbdb! ebfbgbhb6 ib

82 path_to_link, 

83 AppConfig(app_id=app.id, team_id=app.team_id), 

84 ) 

85 

86 result = LinkOutput( 35 ctx2+ , - 7 . / : ; 4 = ? @ [ 8 ] ^ _ ` 5 { | } ~ 9 abbbcbdb! ebfbgbhb6 ib

87 app_id=app.id, 

88 team_id=app.team_id, 

89 path=path_to_link, 

90 app_name=app.name, 

91 config_path=path_to_link / ".fastapicloud" / "cloud.json", 

92 ) 

93 

94 toolkit.success(result, render_output=_render_link_output) 35 ctx2+ , - 7 . / : ; 4 = ? @ [ 8 ] ^ _ ` 5 { | } ~ 9 abbbcbdb! ebfbgbhb6 ib

95 

96 

97def _link_app_interactively( 7 ctx1CDEFGHI

98 toolkit: FastAPIRichToolkit, 

99 *, 

100 path_to_link: Path, 

101 force: bool, 

102) -> None: 

103 if get_app_config(path_to_link) and not force: 63 ctx2a b c jbJ X Q # d e f g kbK Y R $ h i j k lbL Z S % l m n o mbM 0 T ' p q r s nbN 1 U ( t u v w obO 2 V ) x y z A pbP 3 W * B

104 _fail_already_linked_interactive(toolkit) 7 ctx2jbkblbmbnbobpb

105 

106 toolkit.print_title("Link to FastAPI Cloud") 56 ctx1abcJXQ#defgKYR$hijkLZS%lmnoM0T'pqrsN1U(tuvwO2V)xyzAP3W*B

107 toolkit.print_line() 56 ctx1abcJXQ#defgKYR$hijkLZS%lmnoM0T'pqrsN1U(tuvwO2V)xyzAP3W*B

108 

109 with APIClient() as client: 56 ctx1abcJXQ#defgKYR$hijkLZS%lmnoM0T'pqrsN1U(tuvwO2V)xyzAP3W*B

110 with toolkit.progress("Fetching teams...", transient=True) as progress: 56 ctx1abcJXQ#defgKYR$hijkLZS%lmnoM0T'pqrsN1U(tuvwO2V)xyzAP3W*B

111 with client.handle_http_errors( 56 ctx1abcJXQ#defgKYR$hijkLZS%lmnoM0T'pqrsN1U(tuvwO2V)xyzAP3W*B

112 progress, 

113 default_message="Error fetching teams. Please try again later.", 

114 ): 

115 response = client.get("/teams/") 56 ctx1abcJXQ#defgKYR$hijkLZS%lmnoM0T'pqrsN1U(tuvwO2V)xyzAP3W*B

116 response.raise_for_status() 56 ctx1abcJXQ#defgKYR$hijkLZS%lmnoM0T'pqrsN1U(tuvwO2V)xyzAP3W*B

117 teams_data = response.json()["data"] 49 ctx1abcJXQdefgKYRhijkLZSlmnoM0TpqrsN1UtuvwO2VxyzAP3WB

118 

119 if not teams_data: 49 ctx1abcJXQdefgKYRhijkLZSlmnoM0TpqrsN1UtuvwO2VxyzAP3WB

120 toolkit.print( 7 ctx1XYZ0123

121 "[error]No teams found. Please create a team first.[/]", 

122 bullet=False, 

123 ) 

124 raise typer.Exit(1) 7 ctx1XYZ0123

125 

126 team = toolkit.ask( 42 ctx1abcJQdefgKRhijkLSlmnoMTpqrsNUtuvwOVxyzAPWB

127 "Select the team:", 

128 options=[ 

129 Option({"name": t["name"], "value": {"id": t["id"], "name": t["name"]}}) 

130 for t in sorted(teams_data, key=lambda t: t["name"].lower()) 

131 ], 

132 allow_filtering=True, 

133 bullet=False, 

134 ) 

135 

136 toolkit.print_line() 42 ctx1abcJQdefgKRhijkLSlmnoMTpqrsNUtuvwOVxyzAPWB

137 

138 with toolkit.progress("Fetching apps...", transient=True) as progress: 42 ctx1abcJQdefgKRhijkLSlmnoMTpqrsNUtuvwOVxyzAPWB

139 with client.handle_http_errors( 42 ctx1abcJQdefgKRhijkLSlmnoMTpqrsNUtuvwOVxyzAPWB

140 progress, 

141 default_message="Error fetching apps. Please try again later.", 

142 ): 

143 response = client.get("/apps/", params={"team_id": team["id"]}) 42 ctx1abcJQdefgKRhijkLSlmnoMTpqrsNUtuvwOVxyzAPWB

144 response.raise_for_status() 42 ctx1abcJQdefgKRhijkLSlmnoMTpqrsNUtuvwOVxyzAPWB

145 apps_data = response.json()["data"] 35 ctx1abcJdefgKhijkLlmnoMpqrsNtuvwOxyzAPB

146 

147 if not apps_data: 35 ctx1abcJdefgKhijkLlmnoMpqrsNtuvwOxyzAPB

148 toolkit.fail( 7 ctx1JKLMNOP

149 "not_found", 

150 "No apps found in this team.", 

151 hint="Run [bold]fastapi cloud apps create[/] to create and deploy a new app.", 

152 ) 

153 

154 app = toolkit.ask( 28 ctx1abcdefghijklmnopqrstuvwxyzAB

155 "Select the app to link:", 

156 options=[ 

157 Option({"name": a["slug"], "value": {"id": a["id"], "slug": a["slug"]}}) 

158 for a in sorted(apps_data, key=lambda a: a["slug"].lower()) 

159 ], 

160 allow_filtering=True, 

161 bullet=False, 

162 ) 

163 

164 toolkit.print_line() 28 ctx1abcdefghijklmnopqrstuvwxyzAB

165 

166 app_config = AppConfig(app_id=app["id"], team_id=team["id"]) 28 ctx1abcdefghijklmnopqrstuvwxyzAB

167 write_app_config(path_to_link, app_config) 28 ctx1abcdefghijklmnopqrstuvwxyzAB

168 

169 toolkit.print( 28 ctx1abcdefghijklmnopqrstuvwxyzAB

170 f"Successfully linked to app [bold]{app['slug']}[/bold]!", 

171 emoji="🔗", 

172 ) 

173 logger.debug(f"Linked to app: {app['id']} in team: {team['id']}") 28 ctx1abcdefghijklmnopqrstuvwxyzAB

174 

175 

176@cloud_app.command("link") 7 ctx1CDEFGHI

177@apps_app.command("link") 7 ctx1CDEFGHI

178def link_app( 10 ctx2C D ZbE F 0bG H I 1b

179 app_id: Annotated[ 

180 str | None, 

181 typer.Argument( 

182 help="ID of the app to link.", 

183 ), 

184 ] = None, 

185 app_id_option: Annotated[ 

186 str | None, 

187 typer.Option( 

188 "--app-id", 

189 help="ID of the app to link.", 

190 ), 

191 ] = None, 

192 path: Annotated[ 

193 Path | None, 

194 typer.Option( 

195 "--path", 

196 help="Directory to link.", 

197 ), 

198 ] = None, 

199 force: Annotated[ 

200 bool, 

201 typer.Option( 

202 "--force", 

203 help="Replace an existing local app configuration.", 

204 ), 

205 ] = False, 

206 json_output: JsonOutputOption = False, 

207) -> Any: 

208 """ 

209 Link a local directory to an existing FastAPI Cloud app. 

210 """ 

211 identity = Identity() 133 ctx2+ qb, a - 7 . xbLbMbb c jbJ X EbQ # d / rb: e ; 4 = ybNbObf g kbK Y FbR $ h ? sb@ i [ 8 ] zbPbQbj k lbL Z GbS % l ^ tb_ m ` 5 { AbRbSbn o mbM 0 HbT ' p | ub} q ~ 9 abBbTbUbr s nbN 1 IbU ( t bbvbcbu db! ebCbVbWbv w obO 2 JbV ) x fbwbgby hb6 ibDbXbYbz A pbP 3 KbW * B

212 path_to_link = path or Path.cwd() 133 ctx2+ qb, a - 7 . xbLbMbb c jbJ X EbQ # d / rb: e ; 4 = ybNbObf g kbK Y FbR $ h ? sb@ i [ 8 ] zbPbQbj k lbL Z GbS % l ^ tb_ m ` 5 { AbRbSbn o mbM 0 HbT ' p | ub} q ~ 9 abBbTbUbr s nbN 1 IbU ( t bbvbcbu db! ebCbVbWbv w obO 2 JbV ) x fbwbgby hb6 ibDbXbYbz A pbP 3 KbW * B

213 target_app_id = app_id_option or app_id 133 ctx2+ qb, a - 7 . xbLbMbb c jbJ X EbQ # d / rb: e ; 4 = ybNbObf g kbK Y FbR $ h ? sb@ i [ 8 ] zbPbQbj k lbL Z GbS % l ^ tb_ m ` 5 { AbRbSbn o mbM 0 HbT ' p | ub} q ~ 9 abBbTbUbr s nbN 1 IbU ( t bbvbcbu db! ebCbVbWbv w obO 2 JbV ) x fbwbgby hb6 ibDbXbYbz A pbP 3 KbW * B

214 

215 with get_rich_toolkit(json_output=json_output) as toolkit: 133 ctx2+ qb, a - 7 . xbLbMbb c jbJ X EbQ # d / rb: e ; 4 = ybNbObf g kbK Y FbR $ h ? sb@ i [ 8 ] zbPbQbj k lbL Z GbS % l ^ tb_ m ` 5 { AbRbSbn o mbM 0 HbT ' p | ub} q ~ 9 abBbTbUbr s nbN 1 IbU ( t bbvbcbu db! ebCbVbWbv w obO 2 JbV ) x fbwbgby hb6 ibDbXbYbz A pbP 3 KbW * B

216 if not identity.is_logged_in(): 133 ctx2+ qb, a - 7 . xbLbMbb c jbJ X EbQ # d / rb: e ; 4 = ybNbObf g kbK Y FbR $ h ? sb@ i [ 8 ] zbPbQbj k lbL Z GbS % l ^ tb_ m ` 5 { AbRbSbn o mbM 0 HbT ' p | ub} q ~ 9 abBbTbUbr s nbN 1 IbU ( t bbvbcbu db! ebCbVbWbv w obO 2 JbV ) x fbwbgby hb6 ibDbXbYbz A pbP 3 KbW * B

217 if target_app_id is None and not json_output: 14 ctx2LbEbNbFbPbGbRbHbTbIbVbJbXbKb

218 _fail_not_logged_in_interactive(toolkit) 7 ctx2EbFbGbHbIbJbKb

219 

220 toolkit.fail( 7 ctx2LbNbPbRbTbVbXb

221 "not_logged_in", 

222 "No credentials found.", 

223 hint="Run `fastapi cloud login`.", 

224 ) 

225 

226 if app_id is not None and app_id_option is not None and app_id != app_id_option: 119 ctx2+ qb, a - 7 . xbMbb c jbJ X Q # d / rb: e ; 4 = ybObf g kbK Y R $ h ? sb@ i [ 8 ] zbQbj k lbL Z S % l ^ tb_ m ` 5 { AbSbn o mbM 0 T ' p | ub} q ~ 9 abBbUbr s nbN 1 U ( t bbvbcbu db! ebCbWbv w obO 2 V ) x fbwbgby hb6 ibDbYbz A pbP 3 W * B

227 toolkit.fail( 7 ctx2MbObQbSbUbWbYb

228 "invalid_input", 

229 "App ID was provided more than once.", 

230 hint="Pass either APP_ID or --app-id, not both.", 

231 ) 

232 

233 if target_app_id is None: 112 ctx2+ qb, a - 7 . xbb c jbJ X Q # d / rb: e ; 4 = ybf g kbK Y R $ h ? sb@ i [ 8 ] zbj k lbL Z S % l ^ tb_ m ` 5 { Abn o mbM 0 T ' p | ub} q ~ 9 abBbr s nbN 1 U ( t bbvbcbu db! ebCbv w obO 2 V ) x fbwbgby hb6 ibDbz A pbP 3 W * B

234 if json_output: 70 ctx2a xbb c jbJ X Q # d e ybf g kbK Y R $ h i zbj k lbL Z S % l m Abn o mbM 0 T ' p q Bbr s nbN 1 U ( t u Cbv w obO 2 V ) x y Dbz A pbP 3 W * B

235 toolkit.fail( 7 ctx2xbybzbAbBbCbDb

236 "missing_required_input", 

237 "App ID is required.", 

238 hint="Pass an app ID to link an app.", 

239 ) 

240 

241 _link_app_interactively( 63 ctx2a b c jbJ X Q # d e f g kbK Y R $ h i j k lbL Z S % l m n o mbM 0 T ' p q r s nbN 1 U ( t u v w obO 2 V ) x y z A pbP 3 W * B

242 toolkit, 

243 path_to_link=path_to_link, 

244 force=force, 

245 ) 

246 return 28 ctx1abcdefghijklmnopqrstuvwxyzAB

247 

248 _link_app_by_id( 42 ctx2+ qb, - 7 . / rb: ; 4 = ? sb@ [ 8 ] ^ tb_ ` 5 { | ub} ~ 9 abbbvbcbdb! ebfbwbgbhb6 ib

249 toolkit, 

250 app_id=target_app_id, 

251 path_to_link=path_to_link, 

252 force=force, 

253 )