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

1from typing import Annotated, Any 7 ctx1abcdefg

2 

3import typer 7 ctx1abcdefg

4from pydantic import AliasPath, BaseModel, Field 7 ctx1abcdefg

5from rich.table import Table 7 ctx1abcdefg

6from rich.text import Text 7 ctx1abcdefg

7from rich_toolkit import RichToolkit 7 ctx1abcdefg

8 

9from fastapi_cloud_cli.api import APIClient 7 ctx1abcdefg

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

11from fastapi_cloud_cli.commands.integrations.resources._app import resources_app 7 ctx1abcdefg

12from fastapi_cloud_cli.commands.integrations.resources.providers import ( 7 ctx1abcdefg

13 PROVIDER_NAMES, 

14 Provider, 

15) 

16from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail 7 ctx1abcdefg

17from fastapi_cloud_cli.utils.execution import JsonOutputOption 7 ctx1abcdefg

18 

19 

20class ConnectedResource(BaseModel): 7 ctx1abcdefg

21 id: str 7 ctx1abcdefg

22 name: str 7 ctx1abcdefg

23 provider: Provider = Field(validation_alias=AliasPath("provider_metadata", "type")) 7 ctx1abcdefg

24 

25 

26class ConnectedResourcesListAPIResponse(BaseModel): 7 ctx1abcdefg

27 data: list[ConnectedResource] 7 ctx1abcdefg

28 

29 

30class ResourcesListOutput(BaseModel): 7 ctx1abcdefg

31 app_id: str 7 ctx1abcdefg

32 resources: list[ConnectedResource] 7 ctx1abcdefg

33 

34 

35def _get_resources(client: APIClient, *, app_id: str) -> list[ConnectedResource]: 7 ctx1abcdefg

36 response = client.get(f"/apps/{app_id}/connected-resources") 49 ctx1QRvhoJwSTziqLAUVBmrMCWXDjsNEYZxkpKy01FntOG23HluPI

37 response.raise_for_status() 49 ctx1QRvhoJwSTziqLAUVBmrMCWXDjsNEYZxkpKy01FntOG23HluPI

38 

39 return ConnectedResourcesListAPIResponse.model_validate(response.json()).data 42 ctx1QRvhowSTziqAUVBmrCWXDjsEYZxkpy01FntG23HluI

40 

41 

42def _get_resources_table(resources: list[ConnectedResource]) -> Table: 7 ctx1abcdefg

43 table = Table.grid(padding=(0, 2), pad_edge=False) 7 ctx1himjknl

44 table.add_column("Name", no_wrap=True) 7 ctx1himjknl

45 table.add_column("Provider", no_wrap=True) 7 ctx1himjknl

46 table.add_column("Resource ID", no_wrap=True, overflow="ignore") 7 ctx1himjknl

47 table.add_row( 7 ctx1himjknl

48 Text("Name", style="bold"), 

49 Text("Provider", style="bold"), 

50 Text("Resource ID", style="bold"), 

51 ) 

52 table.add_row("", "", "") 7 ctx1himjknl

53 

54 for resource in resources: 7 ctx1himjknl

55 table.add_row( 7 ctx1himjknl

56 Text(resource.name), 

57 Text(PROVIDER_NAMES[resource.provider], style="dim"), 

58 Text(resource.id), 

59 ) 

60 

61 return table 7 ctx1himjknl

62 

63 

64def _render_resources_list_output( 10 ctx1abicdjefgl

65 data: ResourcesListOutput, 

66 toolkit: RichToolkit, 

67) -> None: 

68 toolkit.print_title("connected resources") 14 ctx1hoiqmrjskpntlu

69 toolkit.print_line() 14 ctx1hoiqmrjskpntlu

70 

71 if not data.resources: 14 ctx1hoiqmrjskpntlu

72 toolkit.print("No connected resources found.", bullet=False) 7 ctx1oqrsptu

73 return 7 ctx1oqrsptu

74 

75 toolkit.print(_get_resources_table(data.resources), bullet=False) 7 ctx1himjknl

76 

77 

78@resources_app.command("list", cls=UserCommand) 7 ctx1abcdefg

79def list_resources( 10 ctx1ab#cd$efg%

80 ctx: typer.Context, 

81 app_id: Annotated[ 

82 str | None, 

83 typer.Option( 

84 "--app-id", 

85 help="ID of the app whose connected resources should be listed.", 

86 ), 

87 ] = None, 

88 json_output: JsonOutputOption = False, 

89) -> Any: 

90 """ 

91 List resources connected to an app. 

92 """ 

93 

94 toolkit = get_user_command_context(ctx).toolkit 42 ctx1vho4Jwziq5LABmr6MCDjs7NExkp8KyFnt9OGHlu!PI

95 

96 app_id = resolve_app_id_or_fail(toolkit, app_id=app_id) 42 ctx1vho4Jwziq5LABmr6MCDjs7NExkp8KyFnt9OGHlu!PI

97 

98 with APIClient() as client: 35 ctx1vhoJwziqLABmrMCDjsNExkpKyFntOGHluPI

99 with ( 10 ctx1vhoJwxkpKy

100 toolkit.progress( 

101 title="Fetching connected resources", 

102 transient=True, 

103 ) as progress, 

104 client.handle_http_errors( 

105 progress, 

106 default_message=( 

107 "Error fetching connected resources. Please try again later." 

108 ), 

109 not_found_message="App not found.", 

110 toolkit=toolkit, 

111 ), 

112 ): 

113 resources = _get_resources(client, app_id=app_id) 35 ctx1vhoJwziqLABmrMCDjsNExkpKyFntOGHluPI

114 

115 toolkit.success( 28 ctx1vhowziqABmrCDjsExkpyFntGHluI

116 ResourcesListOutput(app_id=app_id, resources=resources), 

117 render_output=_render_resources_list_output, 

118 )