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

86 statements  

« prev     ^ index     » next       coverage.py v7.16.0, created at 2026-11-09 00:25 +0000

1from typing import Annotated, Any, Literal 1abcdefg

2 

3import typer 1abcdefg

4from pydantic import BaseModel, Field 1abcdefg

5from rich.console import RenderableType 1abcdefg

6from rich.text import Text 1abcdefg

7from rich_toolkit import RichToolkit 1abcdefg

8 

9from fastapi_cloud_cli.api import APIClient 1abcdefg

10from fastapi_cloud_cli.commands._auth import UserCommand, get_user_command_context 1abcdefg

11from fastapi_cloud_cli.commands.integrations.resources._app import resources_app 1abcdefg

12from fastapi_cloud_cli.commands.integrations.resources.providers import PROVIDER_NAMES 1abcdefg

13from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail 1abcdefg

14from fastapi_cloud_cli.utils.cli import get_details_table 1abcdefg

15from fastapi_cloud_cli.utils.dates import format_last_updated 1abcdefg

16from fastapi_cloud_cli.utils.execution import JsonOutputOption 1abcdefg

17 

18DatabaseProvider = Literal["neon", "redis", "supabase"] 1abcdefg

19 

20 

21class DatabaseProviderMetadata(BaseModel): 1abcdefg

22 type: DatabaseProvider 1abcdefg

23 database_name: str 1abcdefg

24 

25 

26class LogfireProviderMetadata(BaseModel): 1abcdefg

27 type: Literal["logfire"] 1abcdefg

28 project_name: str 1abcdefg

29 organization_name: str 1abcdefg

30 

31 

32ProviderMetadata = Annotated[ 1abcdefg

33 DatabaseProviderMetadata | LogfireProviderMetadata, 

34 Field(discriminator="type"), 

35] 

36 

37 

38class EnvironmentVariable(BaseModel): 1abcdefg

39 name: str 1abcdefg

40 

41 

42class ConnectedResourceAPIResponse(BaseModel): 1abcdefg

43 id: str 1abcdefg

44 name: str 1abcdefg

45 provider_metadata: ProviderMetadata 1abcdefg

46 console_url: str 1abcdefg

47 environment_variables: list[EnvironmentVariable] 1abcdefg

48 created_at: str 1abcdefg

49 updated_at: str 1abcdefg

50 

51 

52class ConnectedResourceBase(BaseModel): 1abcdefg

53 id: str 1abcdefg

54 name: str 1abcdefg

55 console_url: str 1abcdefg

56 environment_variables: list[str] 1abcdefg

57 created_at: str 1abcdefg

58 updated_at: str 1abcdefg

59 

60 

61class DatabaseConnectedResource(ConnectedResourceBase): 1abcdefg

62 provider: DatabaseProvider 1abcdefg

63 database_name: str 1abcdefg

64 

65 

66class LogfireConnectedResource(ConnectedResourceBase): 1abcdefg

67 provider: Literal["logfire"] 1abcdefg

68 project_name: str 1abcdefg

69 organization_name: str 1abcdefg

70 

71 

72ConnectedResource = DatabaseConnectedResource | LogfireConnectedResource 1abcdefg

73 

74 

75class ResourceGetOutput(BaseModel): 1abcdefg

76 app_id: str 1abcdefg

77 resource: ConnectedResource 1abcdefg

78 

79 

80def _to_connected_resource( 1abcdefg

81 resource: ConnectedResourceAPIResponse, 

82) -> ConnectedResource: 

83 metadata = resource.provider_metadata 1JKLMxovhNOPQEpyjRSTUFtzkVWXYGqAlZ012Brwi3456HuCm789!IsDn

84 environment_variables = [ 1JKLMxovhNOPQEpyjRSTUFtzkVWXYGqAlZ012Brwi3456HuCm789!IsDn

85 variable.name for variable in resource.environment_variables 

86 ] 

87 

88 match metadata: 1JKLMxovhNOPQEpyjRSTUFtzkVWXYGqAlZ012Brwi3456HuCm789!IsDn

89 case DatabaseProviderMetadata(): 1JKLMxovhNOPQEpyjRSTUFtzkVWXYGqAlZ012Brwi3456HuCm789!IsDn

90 return DatabaseConnectedResource( 1JKLMxoNOPQEpRSTUFtVWXYGqZ012Br3456Hu789!Is

91 id=resource.id, 

92 name=resource.name, 

93 provider=metadata.type, 

94 database_name=metadata.database_name, 

95 console_url=resource.console_url, 

96 environment_variables=environment_variables, 

97 created_at=resource.created_at, 

98 updated_at=resource.updated_at, 

99 ) 

100 case LogfireProviderMetadata(): 1vhyjzkAlwiCmDn

101 return LogfireConnectedResource( 1vhyjzkAlwiCmDn

102 id=resource.id, 

103 name=resource.name, 

104 provider=metadata.type, 

105 project_name=metadata.project_name, 

106 organization_name=metadata.organization_name, 

107 console_url=resource.console_url, 

108 environment_variables=environment_variables, 

109 created_at=resource.created_at, 

110 updated_at=resource.updated_at, 

111 ) 

112 

113 

114def _get_resource( 1abcdefg

115 client: APIClient, 

116 *, 

117 app_id: str, 

118 resource_id: str, 

119) -> ConnectedResource: 

120 response = client.get(f"/apps/{app_id}/connected-resources/{resource_id}") 1JKL+Mxovh#NOP,QEpyj%RST-UFtzk'VWX.YGqAl(Z01/2Brwi$345:6HuCm)789;!IsDn*

121 response.raise_for_status() 1JKL+Mxovh#NOP,QEpyj%RST-UFtzk'VWX.YGqAl(Z01/2Brwi$345:6HuCm)789;!IsDn*

122 

123 resource = ConnectedResourceAPIResponse.model_validate(response.json()) 1JKLMxovhNOPQEpyjRSTUFtzkVWXYGqAlZ012Brwi3456HuCm789!IsDn

124 return _to_connected_resource(resource) 1JKLMxovhNOPQEpyjRSTUFtzkVWXYGqAlZ012Brwi3456HuCm789!IsDn

125 

126 

127def _render_resource_get_output( 1abpcdqefgs

128 data: ResourceGetOutput, 

129 toolkit: RichToolkit, 

130) -> None: 

131 resource = data.resource 1ohpjtkqlriumsn

132 

133 toolkit.print_title("connected resource") 1ohpjtkqlriumsn

134 toolkit.print_line() 1ohpjtkqlriumsn

135 toolkit.print(Text(resource.name, style="bold"), emoji="🔌") 1ohpjtkqlriumsn

136 toolkit.print_line() 1ohpjtkqlriumsn

137 

138 rows: list[tuple[str, RenderableType]] = [ 1ohpjtkqlriumsn

139 ("id", resource.id), 

140 ("provider", PROVIDER_NAMES[resource.provider]), 

141 ] 

142 

143 match resource: 1ohpjtkqlriumsn

144 case DatabaseConnectedResource(): 1ohpjtkqlriumsn

145 rows.append(("database", resource.database_name)) 1optqrus

146 case LogfireConnectedResource(): 1hjklimn

147 rows.extend( 1hjklimn

148 [ 

149 ("project", resource.project_name), 

150 ("organization", resource.organization_name), 

151 ] 

152 ) 

153 

154 rows.extend( 1ohpjtkqlriumsn

155 [ 

156 ( 

157 "console", 

158 Text(resource.console_url, style=f"link {resource.console_url}"), 

159 ), 

160 ( 

161 "environment variables", 

162 Text("\n".join(resource.environment_variables)) 

163 if resource.environment_variables 

164 else Text("-", style="dim"), 

165 ), 

166 ("connected", format_last_updated(resource.created_at)), 

167 ("last updated", format_last_updated(resource.updated_at)), 

168 ] 

169 ) 

170 

171 toolkit.print(get_details_table(rows)) 1ohpjtkqlriumsn

172 

173 

174@resources_app.command("get", cls=UserCommand) 1abcdefg

175def get_resource( 1ab`cd{efg|

176 ctx: typer.Context, 

177 resource_id: Annotated[ 

178 str, 

179 typer.Argument(help="ID of the connected resource to return."), 

180 ], 

181 app_id: Annotated[ 

182 str | None, 

183 typer.Option( 

184 "--app-id", 

185 help="ID of the app that owns the connected resource.", 

186 ), 

187 ] = None, 

188 json_output: JsonOutputOption = False, 

189) -> Any: 

190 """ 

191 Get a resource connected to an app. 

192 """ 

193 

194 toolkit = get_user_command_context(ctx).toolkit 1xovh=#Epyj?%Ftzk@'GqAl[(Brwi]$HuCm^)IsDn_*

195 

196 app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) 1xovh=#Epyj?%Ftzk@'GqAl[(Brwi]$HuCm^)IsDn_*

197 

198 with APIClient() as client: 1xovh#Epyj%Ftzk'GqAl(Brwi$HuCm)IsDn*

199 with ( 1xovh#Brwi$

200 toolkit.progress( 

201 title="Fetching connected resource", 

202 transient=True, 

203 ) as progress, 

204 client.handle_http_errors( 

205 progress, 

206 default_message=( 

207 "Error fetching connected resource. Please try again later." 

208 ), 

209 not_found_message="Connected resource not found.", 

210 toolkit=toolkit, 

211 ), 

212 ): 

213 resource = _get_resource( 1xovh#Epyj%Ftzk'GqAl(Brwi$HuCm)IsDn*

214 client, 

215 app_id=app_id, 

216 resource_id=resource_id, 

217 ) 

218 

219 toolkit.success( 1xovhEpyjFtzkGqAlBrwiHuCmIsDn

220 ResourceGetOutput(app_id=app_id, resource=resource), 

221 render_output=_render_resource_get_output, 

222 )