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

1from typing import Annotated, Any, Literal 7 ctx1abcdefg

2 

3import typer 7 ctx1abcdefg

4from pydantic import BaseModel 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.providers._app import providers_app 7 ctx1abcdefg

12from fastapi_cloud_cli.utils.execution import JsonOutputOption 7 ctx1abcdefg

13from fastapi_cloud_cli.utils.teams import resolve_team_id 7 ctx1abcdefg

14 

15ProviderStatus = Literal["available", "coming_soon", "connected"] 7 ctx1abcdefg

16 

17STATUS_LABELS: dict[ProviderStatus, str] = { 7 ctx1abcdefg

18 "available": "not connected", 

19 "coming_soon": "coming soon", 

20 "connected": "connected", 

21} 

22 

23 

24class ConnectedIntegration(BaseModel): 7 ctx1abcdefg

25 id: str 7 ctx1abcdefg

26 

27 

28class IntegrationProvider(BaseModel): 7 ctx1abcdefg

29 id: str 7 ctx1abcdefg

30 name: str 7 ctx1abcdefg

31 status: ProviderStatus 7 ctx1abcdefg

32 connected_integration: ConnectedIntegration | None = None 7 ctx1abcdefg

33 

34 

35class ProvidersListAPIResponse(BaseModel): 7 ctx1abcdefg

36 data: list[IntegrationProvider] 7 ctx1abcdefg

37 

38 

39class ProvidersListOutput(BaseModel): 7 ctx1abcdefg

40 team_id: str 7 ctx1abcdefg

41 providers: list[IntegrationProvider] 7 ctx1abcdefg

42 

43 

44def _get_providers(client: APIClient, *, team_id: str) -> list[IntegrationProvider]: 7 ctx1abcdefg

45 response = client.get(f"/teams/{team_id}/integrations") 35 ctx1voJhwzqLiABrMmCDsNjExpKkyFtOnGHuPlI

46 response.raise_for_status() 35 ctx1voJhwzqLiABrMmCDsNjExpKkyFtOnGHuPlI

47 

48 return ProvidersListAPIResponse.model_validate(response.json()).data 28 ctx1vohwzqiABrmCDsjExpkyFtnGHulI

49 

50 

51def _get_providers_table(providers: list[IntegrationProvider]) -> Table: 7 ctx1abcdefg

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

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

54 table.add_column("Status", no_wrap=True) 7 ctx1himjknl

55 table.add_column("Integration ID", no_wrap=True, overflow="ignore") 7 ctx1himjknl

56 table.add_row( 7 ctx1himjknl

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

58 Text("Status", style="bold"), 

59 Text("Integration ID", style="bold"), 

60 ) 

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

62 

63 for provider in providers: 7 ctx1himjknl

64 connected_integration = provider.connected_integration 7 ctx1himjknl

65 table.add_row( 7 ctx1himjknl

66 Text(provider.name), 

67 Text( 

68 STATUS_LABELS[provider.status], 

69 style="green" if provider.status == "connected" else "dim", 

70 ), 

71 Text(connected_integration.id if connected_integration else "-"), 

72 ) 

73 

74 return table 7 ctx1himjknl

75 

76 

77def _render_providers_list_output( 10 ctx1abicdjefgl

78 data: ProvidersListOutput, 

79 toolkit: RichToolkit, 

80) -> None: 

81 toolkit.print_title("integrations") 14 ctx1ohqirmsjpktnul

82 toolkit.print_line() 14 ctx1ohqirmsjpktnul

83 

84 if not data.providers: 14 ctx1ohqirmsjpktnul

85 toolkit.print("No integration providers available.", bullet=False) 7 ctx1oqrsptu

86 return 7 ctx1oqrsptu

87 

88 toolkit.print(_get_providers_table(data.providers), bullet=False) 7 ctx1himjknl

89 

90 

91@providers_app.command("list", cls=UserCommand) 7 ctx1abcdefg

92def list_providers( 10 ctx1ab4cd5efg6

93 ctx: typer.Context, 

94 team_id: Annotated[ 

95 str | None, 

96 typer.Option( 

97 "--team-id", 

98 help=( 

99 "ID of the team whose integration providers should be listed. " 

100 "Defaults to the linked app's team." 

101 ), 

102 ), 

103 ] = None, 

104 json_output: JsonOutputOption = False, 

105) -> Any: 

106 """ 

107 List integration providers and their connection status for a team. 

108 """ 

109 

110 toolkit = get_user_command_context(ctx).toolkit 49 ctx1voQJhRwzqSLiTABrUMmVCDsWNjXExpYKkZyFt0On1GHu2Pl3I

111 

112 with APIClient() as client: 49 ctx1voQJhRwzqSLiTABrUMmVCDsWNjXExpYKkZyFt0On1GHu2Pl3I

113 team_id = resolve_team_id( 49 ctx1voQJhRwzqSLiTABrUMmVCDsWNjXExpYKkZyFt0On1GHu2Pl3I

114 toolkit, 

115 client, 

116 team_id=team_id, 

117 empty_hint="Create a team before listing integration providers.", 

118 ) 

119 

120 with ( 10 ctx1voJhwxpKky

121 toolkit.progress( 

122 title="Fetching integration providers", 

123 transient=True, 

124 ) as progress, 

125 client.handle_http_errors( 

126 progress, 

127 default_message=( 

128 "Error fetching integration providers. Please try again later." 

129 ), 

130 not_found_message="Team not found.", 

131 toolkit=toolkit, 

132 ), 

133 ): 

134 providers = _get_providers(client, team_id=team_id) 35 ctx1voJhwzqLiABrMmCDsNjExpKkyFtOnGHuPlI

135 

136 toolkit.success( 28 ctx1vohwzqiABrmCDsjExpkyFtnGHulI

137 ProvidersListOutput(team_id=team_id, providers=providers), 

138 render_output=_render_providers_list_output, 

139 )