Coverage for src/fastapi_cloud_cli/commands/integrations/resources/connect.py: 100%

1import logging 7 ctx1fghijkl

2from typing import Annotated, Any 7 ctx1fghijkl

3 

4import typer 7 ctx1fghijkl

5from pydantic import BaseModel, Field 7 ctx1fghijkl

6from rich.text import Text 7 ctx1fghijkl

7from rich_toolkit import RichToolkit 7 ctx1fghijkl

8 

9from fastapi_cloud_cli.api import APIClient 7 ctx1fghijkl

10from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context 7 ctx1fghijkl

11from fastapi_cloud_cli.commands.apps.list import ( 7 ctx1fghijkl

12 App, 

13 _get_app, 

14 _get_app_dashboard_url, 

15 _get_team, 

16) 

17from fastapi_cloud_cli.commands.integrations.resources._app import resources_app 7 ctx1fghijkl

18from fastapi_cloud_cli.config import Settings 7 ctx1fghijkl

19from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail 7 ctx1fghijkl

20from fastapi_cloud_cli.utils.execution import JsonOutputOption 7 ctx1fghijkl

21 

22logger = logging.getLogger(__name__) 7 ctx1fghijkl

23 

24 

25class ResourceConnectOutput(BaseModel): 7 ctx1fghijkl

26 app_id: str 7 ctx1fghijkl

27 connect_url: str 7 ctx1fghijkl

28 app_name: Annotated[str, Field(exclude=True)] 7 ctx1fghijkl

29 browser_opened: Annotated[bool, Field(exclude=True)] 7 ctx1fghijkl

30 app_id_was_provided: Annotated[bool, Field(exclude=True)] 7 ctx1fghijkl

31 

32 

33def _get_resource_connect_url( 7 ctx1fghijkl

34 app: App, 

35 *, 

36 team_slug: str, 

37 settings: Settings, 

38) -> str: 

39 app_dashboard_url = _get_app_dashboard_url( 21 ctx1vmaxqcyrnzsdwobAtpBue

40 app, 

41 team_slug=team_slug, 

42 settings=settings, 

43 ) 

44 return f"{app_dashboard_url}/integrations/connect" 21 ctx1vmaxqcyrnzsdwobAtpBue

45 

46 

47def _render_resource_connect_output( 10 ctx1fgchidjkle

48 data: ResourceConnectOutput, 

49 toolkit: RichToolkit, 

50) -> None: 

51 toolkit.print_title("connect resource") 14 ctx1maqcrnsdobtpue

52 toolkit.print_line() 14 ctx1maqcrnsdobtpue

53 

54 if data.browser_opened: 14 ctx1maqcrnsdobtpue

55 toolkit.print( 7 ctx1acndbpe

56 f"Opened the integration setup for [bold]{data.app_name}[/bold] in your browser.", 

57 emoji="🔌", 

58 ) 

59 else: 

60 toolkit.print( 7 ctx1mqrsotu

61 f"Open the integration setup for [bold]{data.app_name}[/bold] in your browser:", 

62 emoji="🔌", 

63 ) 

64 

65 toolkit.print(Text(data.connect_url, style=f"link {data.connect_url}")) 14 ctx1maqcrnsdobtpue

66 toolkit.print_line() 14 ctx1maqcrnsdobtpue

67 

68 list_command = "fastapi cloud integrations resources list" 14 ctx1maqcrnsdobtpue

69 if data.app_id_was_provided: 14 ctx1maqcrnsdobtpue

70 list_command = f"{list_command} --app-id {data.app_id}" 7 ctx1mqrsotu

71 

72 toolkit.print( 14 ctx1maqcrnsdobtpue

73 f"When the connection is complete, run [bold]{list_command}[/bold] to view it.", 

74 emoji="💡", 

75 ) 

76 

77 

78@resources_app.command("connect", cls=UserCommand) 7 ctx1fghijkl

79def connect_resource( 10 ctx1fgQhiRjklS

80 ctx: typer.Context, 

81 app_id: Annotated[ 

82 str | None, 

83 typer.Option( 

84 "--app-id", 

85 help="ID of the app to connect a provider resource to.", 

86 ), 

87 ] = None, 

88 no_open: Annotated[ 

89 bool, 

90 typer.Option( 

91 "--no-open", 

92 help="Do not open the browser automatically.", 

93 ), 

94 ] = False, 

95 json_output: JsonOutputOption = False, 

96) -> Any: 

97 """ 

98 Open FastAPI Cloud to connect a provider resource to an app. 

99 """ 

100 

101 toolkit = get_user_command_context(ctx).toolkit 35 ctx1JCvmaKExqcLFyrnMGzsdNDwobOHAtpPIBue

102 

103 app_id_was_provided = app_id is not None 35 ctx1JCvmaKExqcLFyrnMGzsdNDwobOHAtpPIBue

104 app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) 35 ctx1JCvmaKExqcLFyrnMGzsdNDwobOHAtpPIBue

105 

106 with APIClient() as client: 28 ctx1CvmaExqcFyrnGzsdDwobHAtpIBue

107 with ( 8 ctx1CvmaDwob

108 toolkit.progress(title="Fetching app", transient=True) as progress, 

109 client.handle_http_errors( 

110 progress, 

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

112 not_found_message="App not found.", 

113 toolkit=toolkit, 

114 ), 

115 ): 

116 app = _get_app(client, app_id) 28 ctx1CvmaExqcFyrnGzsdDwobHAtpIBue

117 

118 with ( 6 ctx1vmawob

119 toolkit.progress(title="Fetching team", transient=True) as progress, 

120 client.handle_http_errors( 

121 progress, 

122 default_message="Error fetching team. Please try again later.", 

123 not_found_message="Team not found.", 

124 toolkit=toolkit, 

125 ), 

126 ): 

127 team = _get_team(client, app.team_id) 21 ctx1vmaxqcyrnzsdwobAtpBue

128 

129 connect_url = _get_resource_connect_url( 21 ctx1vmaxqcyrnzsdwobAtpBue

130 app, 

131 team_slug=team.slug, 

132 settings=Settings.get(), 

133 ) 

134 browser_opened = False 21 ctx1vmaxqcyrnzsdwobAtpBue

135 

136 if not json_output and not no_open: 21 ctx1vmaxqcyrnzsdwobAtpBue

137 launch_result = typer.launch(connect_url) 7 ctx1acndbpe

138 logger.debug("Launch command result: %s", launch_result) 7 ctx1acndbpe

139 browser_opened = launch_result == 0 7 ctx1acndbpe

140 

141 toolkit.success( 21 ctx1vmaxqcyrnzsdwobAtpBue

142 ResourceConnectOutput( 

143 app_id=app.id, 

144 app_name=app.name, 

145 connect_url=connect_url, 

146 browser_opened=browser_opened, 

147 app_id_was_provided=app_id_was_provided, 

148 ), 

149 render_output=_render_resource_connect_output, 

150 )