Skip to content

♻️ Use handle_http_errors in login command #45

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
patrick91 merged 1 commit into from
May 6, 2025
Merged

♻️ Use handle_http_errors in login command #45

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 15 additions & 37 deletions src/fastapi_cloud_cli/commands/login.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import logging
import time
from typing import Any, Optional
from typing import Any

import httpx
import typer
Expand All @@ -9,7 +9,7 @@
from fastapi_cloud_cli.config import settings
from fastapi_cloud_cli.utils.api import APIClient
from fastapi_cloud_cli.utils.auth import AuthConfig, write_auth_config
from fastapi_cloud_cli.utils.cli import get_rich_toolkit
from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors

logger = logging.getLogger(__name__)

Expand All @@ -28,25 +28,17 @@ class TokenResponse(BaseModel):

def _start_device_authorization(
client: httpx.Client,
) -> Optional[AuthorizationData]:
try:
response = client.post(
"/login/device/authorization", data={"client_id": settings.client_id}
)

response.raise_for_status()
) -> AuthorizationData:
response = client.post(
"/login/device/authorization", data={"client_id": settings.client_id}
)

except httpx.HTTPError as e:
logger.debug("Error: %s", e)

return None
response.raise_for_status()

return AuthorizationData.model_validate(response.json())


def _fetch_access_token(
client: httpx.Client, device_code: str, interval: int
) -> Optional[str]:
def _fetch_access_token(client: httpx.Client, device_code: str, interval: int) -> str:
while True:
response = client.post(
"/login/device/token",
Expand All @@ -58,15 +50,13 @@ def _fetch_access_token(
)

if response.status_code not in (200, 400):
logger.debug("Error: %s", response.json())
return None
response.raise_for_status()

if response.status_code == 400:
data = response.json()

if data.get("error") != "authorization_pending":
logger.debug("Error: %s", data)
return None
response.raise_for_status()

if response.status_code == 200:
break
Expand All @@ -88,14 +78,8 @@ def login() -> Any:
toolkit.print_line()

with toolkit.progress("Starting authorization") as progress:
authorization_data = _start_device_authorization(client)

if authorization_data is None:
progress.set_error(
"Something went wrong while contacting the FastAPI Cloud server. Please try again later."
)

raise typer.Exit(1)
with handle_http_errors(progress):
authorization_data = _start_device_authorization(client)

url = authorization_data.verification_uri_complete

Expand All @@ -106,17 +90,11 @@ def login() -> Any:
with toolkit.progress("Waiting for user to authorize...") as progress:
typer.launch(url)

access_token = _fetch_access_token(
client, authorization_data.device_code, authorization_data.interval
)

if access_token is None:
progress.set_error(
"Something went wrong while contacting the FastAPI Cloud server. Please try again later."
with handle_http_errors(progress):
access_token = _fetch_access_token(
client, authorization_data.device_code, authorization_data.interval
)

raise typer.Exit(1)

write_auth_config(AuthConfig(access_token=access_token))

progress.log("Now you are logged in! 🚀")
Loading