Coverage for src/fastapi_cloud_cli/commands/_flow.py: 100%

85 statements  

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

1import logging 1abcdefg

2import time 1abcdefg

3 

4import httpx 1abcdefg

5from pydantic import BaseModel 1abcdefg

6from rich_toolkit import RichToolkit 1abcdefg

7from rich_toolkit.progress import Progress 1abcdefg

8 

9from fastapi_cloud_cli.api import APIClient 1abcdefg

10from fastapi_cloud_cli.config import Settings 1abcdefg

11from fastapi_cloud_cli.utils.auth import AuthConfig, AuthMode, write_auth_config 1abcdefg

12from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit 1abcdefg

13 

14logger = logging.getLogger(__name__) 1abcdefg

15 

16DEFAULT_LOGIN_TIMEOUT_SECONDS = 300 1abcdefg

17 

18 

19class AuthorizationData(BaseModel): 1abcdefg

20 user_code: str 1abcdefg

21 device_code: str 1abcdefg

22 verification_uri: str 1abcdefg

23 verification_uri_complete: str 1abcdefg

24 interval: int = 5 1abcdefg

25 

26 

27class TokenResponse(BaseModel): 1abcdefg

28 access_token: str 1abcdefg

29 

30 

31class LoginOutput(BaseModel): 1abcdefg

32 authenticated: bool 1abcdefg

33 auth_mode: AuthMode 1abcdefg

34 

35 

36class DeviceAuthorizationOutput(BaseModel): 1abcdefg

37 verification_uri: str 1abcdefg

38 verification_uri_complete: str 1abcdefg

39 user_code: str 1abcdefg

40 device_code: str 1abcdefg

41 interval: int 1abcdefg

42 

43 

44class LoginTimeoutError(Exception): 1abcdefg

45 pass 1abcdefg

46 

47 

48class LoginDeniedError(Exception): 1abcdefg

49 pass 1abcdefg

50 

51 

52def render_login_output(data: LoginOutput, toolkit: RichToolkit) -> None: 1abhcdiefgj

53 toolkit.print("Now you are logged in! 🚀") 1r+stuvh,wxyzA-BCDEi.FGHIJ/KLMNO:PQRSj;TUVW

54 

55 

56def device_authorization_output( 1abcdefg

57 authorization_data: AuthorizationData, 

58) -> DeviceAuthorizationOutput: 

59 return DeviceAuthorizationOutput( 2LbMbNbObPbQbRbSbTbUbVbWbXbYbZb0b1b2b3b4b5b

60 verification_uri=authorization_data.verification_uri, 

61 verification_uri_complete=authorization_data.verification_uri_complete, 

62 user_code=authorization_data.user_code, 

63 device_code=authorization_data.device_code, 

64 interval=authorization_data.interval, 

65 ) 

66 

67 

68def start_device_authorization( 1abcdefg

69 client: httpx.Client, 

70) -> AuthorizationData: 

71 settings = Settings.get() 2r Lbs Mbt u v 4 6bNbh Obw Pbx y z 5 7bQbA RbB SbC D E 6 8bTbi UbF VbG H I 7 9bWbJ XbK YbL M N 8 !bZbO 0bP 1bQ R S 9 #b2bj 3bT 4bU V W ! $b5b

72 

73 response = client.post( 2r Lbs Mbt u v 4 6bNbh Obw Pbx y z 5 7bQbA RbB SbC D E 6 8bTbi UbF VbG H I 7 9bWbJ XbK YbL M N 8 !bZbO 0bP 1bQ R S 9 #b2bj 3bT 4bU V W ! $b5b

74 "/login/device/authorization", data={"client_id": settings.client_id} 

75 ) 

76 logger.debug(f"Device authorization response status code: {response.status_code}") 2r Lbs Mbt u v 4 6bNbh Obw Pbx y z 5 7bQbA RbB SbC D E 6 8bTbi UbF VbG H I 7 9bWbJ XbK YbL M N 8 !bZbO 0bP 1bQ R S 9 #b2bj 3bT 4bU V W ! $b5b

77 

78 response.raise_for_status() 2r Lbs Mbt u v 4 6bNbh Obw Pbx y z 5 7bQbA RbB SbC D E 6 8bTbi UbF VbG H I 7 9bWbJ XbK YbL M N 8 !bZbO 0bP 1bQ R S 9 #b2bj 3bT 4bU V W ! $b5b

79 

80 return AuthorizationData.model_validate_json(response.text) 2r Lbs Mbt u v 4 Nbh Obw Pbx y z 5 QbA RbB SbC D E 6 Tbi UbF VbG H I 7 WbJ XbK YbL M N 8 ZbO 0bP 1bQ R S 9 2bj 3bT 4bU V W ! 5b

81 

82 

83def fetch_access_token( 1abcdefg

84 client: httpx.Client, 

85 device_code: str, 

86 interval: int, 

87 timeout: int = DEFAULT_LOGIN_TIMEOUT_SECONDS, 

88) -> str: 

89 settings = Settings.get() 2r = k X ? + s # xbjbqbcbt u v 4 h @ l Y [ , w $ ybkbrbdbx y z 5 A ] m Z ^ - B % zblbsbebC D E 6 i _ n 0 ` . F ' AbmbtbfbG H I 7 J { o 1 | / K ( BbnbubgbL M N 8 O } p 2 ~ : P ) CbobvbhbQ R S 9 j abq 3 bb; T * DbpbwbibU V W !

90 start = time.monotonic() 2r = k X ? + s # xbjbqbcbt u v 4 h @ l Y [ , w $ ybkbrbdbx y z 5 A ] m Z ^ - B % zblbsbebC D E 6 i _ n 0 ` . F ' AbmbtbfbG H I 7 J { o 1 | / K ( BbnbubgbL M N 8 O } p 2 ~ : P ) CbobvbhbQ R S 9 j abq 3 bb; T * DbpbwbibU V W !

91 

92 logger.debug("Starting to poll for access token") 2r = k X ? + s # xbjbqbcbt u v 4 h @ l Y [ , w $ ybkbrbdbx y z 5 A ] m Z ^ - B % zblbsbebC D E 6 i _ n 0 ` . F ' AbmbtbfbG H I 7 J { o 1 | / K ( BbnbubgbL M N 8 O } p 2 ~ : P ) CbobvbhbQ R S 9 j abq 3 bb; T * DbpbwbibU V W !

93 while True: 2r = k X ? + s # xbjbqbcbt u v 4 h @ l Y [ , w $ ybkbrbdbx y z 5 A ] m Z ^ - B % zblbsbebC D E 6 i _ n 0 ` . F ' AbmbtbfbG H I 7 J { o 1 | / K ( BbnbubgbL M N 8 O } p 2 ~ : P ) CbobvbhbQ R S 9 j abq 3 bb; T * DbpbwbibU V W !

94 response = client.post( 2r = k X ? + s # xbjbqbcbt u v 4 h @ l Y [ , w $ ybkbrbdbx y z 5 A ] m Z ^ - B % zblbsbebC D E 6 i _ n 0 ` . F ' AbmbtbfbG H I 7 J { o 1 | / K ( BbnbubgbL M N 8 O } p 2 ~ : P ) CbobvbhbQ R S 9 j abq 3 bb; T * DbpbwbibU V W !

95 "/login/device/token", 

96 data={ 

97 "device_code": device_code, 

98 "client_id": settings.client_id, 

99 "grant_type": "urn:ietf:params:oauth:grant-type:device_code", 

100 }, 

101 ) 

102 logger.debug(f"Token response status code: {response.status_code}") 2r = k X ? + s # xbjbqbcbt u v 4 h @ l Y [ , w $ ybkbrbdbx y z 5 A ] m Z ^ - B % zblbsbebC D E 6 i _ n 0 ` . F ' AbmbtbfbG H I 7 J { o 1 | / K ( BbnbubgbL M N 8 O } p 2 ~ : P ) CbobvbhbQ R S 9 j abq 3 bb; T * DbpbwbibU V W !

103 

104 if response.status_code not in (200, 400): 2r = k X ? + s # xbjbqbcbt u v 4 h @ l Y [ , w $ ybkbrbdbx y z 5 A ] m Z ^ - B % zblbsbebC D E 6 i _ n 0 ` . F ' AbmbtbfbG H I 7 J { o 1 | / K ( BbnbubgbL M N 8 O } p 2 ~ : P ) CbobvbhbQ R S 9 j abq 3 bb; T * DbpbwbibU V W !

105 response.raise_for_status() 2xbybzbAbBbCbDb

106 

107 if response.status_code == 400: 2r = k X ? + s # jbqbcbt u v 4 h @ l Y [ , w $ kbrbdbx y z 5 A ] m Z ^ - B % lbsbebC D E 6 i _ n 0 ` . F ' mbtbfbG H I 7 J { o 1 | / K ( nbubgbL M N 8 O } p 2 ~ : P ) obvbhbQ R S 9 j abq 3 bb; T * pbwbibU V W !

108 data = response.json() 2k X ? # jbcb4 l Y [ $ kbdb5 m Z ^ % lbeb6 n 0 ` ' mbfb7 o 1 | ( nbgb8 p 2 ~ ) obhb9 q 3 bb* pbib!

109 error = data.get("error") 2k X ? # jbcb4 l Y [ $ kbdb5 m Z ^ % lbeb6 n 0 ` ' mbfb7 o 1 | ( nbgb8 p 2 ~ ) obhb9 q 3 bb* pbib!

110 logger.debug(f"Token response error: {error}") 2k X ? # jbcb4 l Y [ $ kbdb5 m Z ^ % lbeb6 n 0 ` ' mbfb7 o 1 | ( nbgb8 p 2 ~ ) obhb9 q 3 bb* pbib!

111 

112 if error == "access_denied": 2k X ? # jbcb4 l Y [ $ kbdb5 m Z ^ % lbeb6 n 0 ` ' mbfb7 o 1 | ( nbgb8 p 2 ~ ) obhb9 q 3 bb* pbib!

113 raise LoginDeniedError 2k ? 4 l [ 5 m ^ 6 n ` 7 o | 8 p ~ 9 q bb!

114 

115 if error != "authorization_pending": 2k X # jbcbl Y $ kbdbm Z % lbebn 0 ' mbfbo 1 ( nbgbp 2 ) obhbq 3 * pbib

116 response.raise_for_status() 2jbkblbmbnbobpb

117 

118 if response.status_code == 200: 2r = k X + s # qbcbt u v h @ l Y , w $ rbdbx y z A ] m Z - B % sbebC D E i _ n 0 . F ' tbfbG H I J { o 1 / K ( ubgbL M N O } p 2 : P ) vbhbQ R S j abq 3 ; T * wbibU V W

119 break 2r = + s # qbt u v h @ , w $ rbx y z A ] - B % sbC D E i _ . F ' tbG H I J { / K ( ubL M N O } : P ) vbQ R S j ab; T * wbU V W

120 

121 remaining = timeout - (time.monotonic() - start) 2k X # cbl Y $ dbm Z % ebn 0 ' fbo 1 ( gbp 2 ) hbq 3 * ib

122 if remaining <= 0: 2k X # cbl Y $ dbm Z % ebn 0 ' fbo 1 ( gbp 2 ) hbq 3 * ib

123 raise LoginTimeoutError 2X cbY dbZ eb0 fb1 gb2 hb3 ib

124 

125 sleep_for = min(interval, remaining) 1k#l$m%n'o(p)q*

126 

127 logger.debug(f"Sleeping for {sleep_for} seconds before retrying...") 1k#l$m%n'o(p)q*

128 time.sleep(sleep_for) 1k#l$m%n'o(p)q*

129 

130 response_data = TokenResponse.model_validate_json(response.text) 2r = + s # qbt u v h @ , w $ rbx y z A ] - B % sbC D E i _ . F ' tbG H I J { / K ( ubL M N O } : P ) vbQ R S j ab; T * wbU V W

131 logger.debug("Access token received successfully.") 2r = + s # qbt u v h @ , w $ rbx y z A ] - B % sbC D E i _ . F ' tbG H I J { / K ( ubL M N O } : P ) vbQ R S j ab; T * wbU V W

132 

133 return response_data.access_token 2r = + s # qbt u v h @ , w $ rbx y z A ] - B % sbC D E i _ . F ' tbG H I J { / K ( ubL M N O } : P ) vbQ R S j ab; T * wbU V W

134 

135 

136def complete_device_login( 1abcdefg

137 *, 

138 client: APIClient, 

139 progress: Progress, 

140 toolkit: FastAPIRichToolkit, 

141 device_code: str, 

142 interval: int, 

143 timeout: int, 

144 cancel_hint: str, 

145) -> LoginOutput: 

146 try: 2r = k EbX ? + s t u v 4 h @ l FbY [ , w x y z 5 A ] m GbZ ^ - B C D E 6 i _ n Hb0 ` . F G H I 7 J { o Ib1 | / K L M N 8 O } p Jb2 ~ : P Q R S 9 j abq Kb3 bb; T U V W !

147 with client.handle_http_errors(progress, toolkit=toolkit): 2r = k EbX ? + s t u v 4 h @ l FbY [ , w x y z 5 A ] m GbZ ^ - B C D E 6 i _ n Hb0 ` . F G H I 7 J { o Ib1 | / K L M N 8 O } p Jb2 ~ : P Q R S 9 j abq Kb3 bb; T U V W !

148 access_token = fetch_access_token(client, device_code, interval, timeout) 2r = k EbX ? + s t u v 4 h @ l FbY [ , w x y z 5 A ] m GbZ ^ - B C D E 6 i _ n Hb0 ` . F G H I 7 J { o Ib1 | / K L M N 8 O } p Jb2 ~ : P Q R S 9 j abq Kb3 bb; T U V W !

149 except LoginDeniedError: 2k EbX ? 4 l FbY [ 5 m GbZ ^ 6 n Hb0 ` 7 o Ib1 | 8 p Jb2 ~ 9 q Kb3 bb!

150 toolkit.fail( 2k ? 4 l [ 5 m ^ 6 n ` 7 o | 8 p ~ 9 q bb!

151 "permission_denied", 

152 "Login authorization was declined.", 

153 hint="Run `fastapi cloud login` again and approve the authorization request.", 

154 ) 

155 except LoginTimeoutError: 2EbX FbY GbZ Hb0 Ib1 Jb2 Kb3

156 message = "Login timed out before authorization completed." 1XYZ0123

157 toolkit.fail( 1XYZ0123

158 "timeout", 

159 message, 

160 hint="Try again with a longer --timeout value.", 

161 ) 

162 except KeyboardInterrupt: 2EbFbGbHbIbJbKb

163 message = "Login cancelled before authorization completed." 2EbFbGbHbIbJbKb

164 toolkit.fail( 2EbFbGbHbIbJbKb

165 "cancelled", 

166 message, 

167 hint=cancel_hint, 

168 ) 

169 

170 write_auth_config(AuthConfig(access_token=access_token)) 2r = + s t u v h @ , w x y z A ] - B C D E i _ . F G H I J { / K L M N O } : P Q R S j ab; T U V W

171 

172 return LoginOutput(authenticated=True, auth_mode="user") 2r = + s t u v h @ , w x y z A ] - B C D E i _ . F G H I J { / K L M N O } : P Q R S j ab; T U V W