Add login command

Author: patrick91

pyproject.toml

@@ -35,6 +35,7 @@ classifiers = [
 dependencies = [
     "typer >= 0.12.3",
     "rich-toolkit >= 0.6.3",
+    "pydantic >= 1.6.1",
 ]
 
 [project.optional-dependencies]

requirements-tests.txt

@@ -4,6 +4,7 @@ pytest >=4.4.0,<9.0.0
 coverage[toml] >=6.2,<8.0
 mypy ==1.10.0
 ruff ==0.4.5
+respx ==0.21.1
 # Needed explicitly by fastapi-cli-slim
 fastapi-slim
 uvicorn

src/fastapi_cli/cli.py

@@ -12,13 +12,16 @@
 from fastapi_cli.exceptions import FastAPICLIException
 
 from . import __version__
+from .commands.login import login
+from .config import settings
 from .logging import setup_logging
 
 app = typer.Typer(rich_markup_mode="rich")
 
 setup_logging()
 logger = getLogger(__name__)
 
+
 try:
     import uvicorn
 except ImportError:  # pragma: no cover
@@ -39,6 +42,7 @@ def callback(
             "--version", help="Show the version and exit.", callback=version_callback
         ),
     ] = None,
+    base_api_url: Union[str, None] = typer.Option(None, help="Base URL of the API"),
 ) -> None:
     """
     FastAPI CLI - The [bold]fastapi[/bold] command line app. 😎
@@ -47,6 +51,8 @@ def callback(
 
     Read more in the docs: [link]https://fastapi.tiangolo.com/fastapi-cli/[/link].
     """
+    if base_api_url:
+        settings.base_api_url = base_api_url
 
 
 def _run(
@@ -273,5 +279,9 @@ def run(
     )
 
 
+# Additional commands
+app.command()(login)
+
+
 def main() -> None:
     app()

src/fastapi_cli/commands/__init__.py

@@ -0,0 +1,122 @@
+import logging
+import time
+from typing import Any, Optional
+
+import httpx
+import typer
+from pydantic import BaseModel
+
+from fastapi_cli.config import settings
+from fastapi_cli.utils.api import APIClient
+from fastapi_cli.utils.auth import AuthConfig, write_auth_config
+from fastapi_cli.utils.cli import get_rich_toolkit
+
+logger = logging.getLogger(__name__)
+
+
+class AuthorizationData(BaseModel):
+    user_code: str
+    device_code: str
+    verification_uri: str
+    verification_uri_complete: str
+    interval: int = 5
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+
+
+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()
+
+    except httpx.HTTPError as e:
+        logger.debug("Error: %s", e)
+
+        return None
+
+    return AuthorizationData.model_validate(response.json())
+
+
+def _fetch_access_token(
+    client: httpx.Client, device_code: str, interval: int
+) -> Optional[str]:
+    while True:
+        response = client.post(
+            "/login/device/token",
+            data={
+                "device_code": device_code,
+                "client_id": settings.client_id,
+                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+            },
+        )
+
+        if response.status_code not in (200, 400):
+            logger.debug("Error: %s", response.json())
+            return None
+
+        if response.status_code == 400:
+            data = response.json()
+
+            if data.get("error") != "authorization_pending":
+                logger.debug("Error: %s", data)
+                return None
+
+        if response.status_code == 200:
+            break
+
+        time.sleep(interval)
+
+    response_data = TokenResponse.model_validate(response.json())
+
+    return response_data.access_token
+
+
+def login() -> Any:
+    """
+    Login to FastAPI Cloud. 🚀
+    """
+    with get_rich_toolkit() as toolkit, APIClient() as client:
+        toolkit.print_title("Login to FastAPI Cloud", tag="FastAPI")
+
+        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)
+
+            url = authorization_data.verification_uri_complete
+
+            progress.log(f"Opening {url}")
+
+        toolkit.print_line()
+
+        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."
+                )
+
+                raise typer.Exit(1)
+
+            write_auth_config(AuthConfig(access_token=access_token))
+
+            progress.log("Now you are logged in! 🚀")

src/fastapi_cli/commands/login.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel
+
+
+class Settings(BaseModel):
+    base_api_url: str = "http://localhost:8000/api/v1"
+    client_id: str = "fastapi-cli"
+
+
+settings = Settings()

src/fastapi_cli/config.py

@@ -0,0 +1,14 @@
+import httpx
+
+from fastapi_cli.config import settings
+
+from .auth import get_auth_token
+
+
+class APIClient(httpx.Client):
+    def __init__(self) -> None:
+        token = get_auth_token()
+
+        super().__init__(
+            base_url=settings.base_api_url, headers={"Authorization": f"Bearer {token}"}
+        )

src/fastapi_cli/utils/api.py

@@ -0,0 +1,33 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+from .config import get_auth_path
+
+
+class AuthConfig(BaseModel):
+    access_token: str
+
+
+def write_auth_config(auth_data: AuthConfig) -> None:
+    auth_path = get_auth_path()
+
+    auth_path.write_text(auth_data.model_dump_json(), encoding="utf-8")
+
+
+def read_auth_config() -> Optional[AuthConfig]:
+    auth_path = get_auth_path()
+
+    if not auth_path.exists():
+        return None
+
+    return AuthConfig.model_validate_json(auth_path.read_text(encoding="utf-8"))
+
+
+def get_auth_token() -> Optional[str]:
+    auth_data = read_auth_config()
+
+    if auth_data is None:
+        return None
+
+    return auth_data.access_token

src/fastapi_cli/utils/auth.py

@@ -0,0 +1,14 @@
+from pathlib import Path
+
+import typer
+
+
+def get_config_folder() -> Path:
+    return Path(typer.get_app_dir("fastapi-cli"))
+
+
+def get_auth_path() -> Path:
+    auth_path = get_config_folder() / "auth.json"
+    auth_path.parent.mkdir(parents=True, exist_ok=True)
+
+    return auth_path

src/fastapi_cli/utils/config.py

@@ -0,0 +1,64 @@
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+import respx
+from fastapi_cli.cli import app
+from fastapi_cli.config import settings
+from httpx import Response
+from typer.testing import CliRunner
+
+runner = CliRunner()
+
+assets_path = Path(__file__).parent / "assets"
+
+
+@pytest.mark.respx(base_url=settings.base_api_url)
+def test_shows_a_message_if_something_is_wrong(respx_mock: respx.MockRouter) -> None:
+    with patch("fastapi_cli.commands.login.typer.launch") as mock_open:
+        respx_mock.post(
+            "/login/device/authorization", data={"client_id": settings.client_id}
+        ).mock(return_value=Response(500))
+
+        result = runner.invoke(app, ["login"])
+
+        assert result.exit_code == 1
+        assert (
+            "Something went wrong while contacting the FastAPI Cloud server."
+            in result.output
+        )
+
+        assert not mock_open.called
+
+
+@pytest.mark.respx(base_url=settings.base_api_url)
+def test_full_login(respx_mock: respx.MockRouter) -> None:
+    with patch("fastapi_cli.commands.login.typer.launch") as mock_open:
+        respx_mock.post(
+            "/login/device/authorization", data={"client_id": settings.client_id}
+        ).mock(
+            return_value=Response(
+                200,
+                json={
+                    "verification_uri_complete": "http://test.com",
+                    "verification_uri": "http://test.com",
+                    "user_code": "1234",
+                    "device_code": "5678",
+                },
+            )
+        )
+        respx_mock.post(
+            "/login/device/token",
+            data={
+                "device_code": "5678",
+                "client_id": settings.client_id,
+                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+            },
+        ).mock(return_value=Response(200, json={"access_token": "1234"}))
+
+        result = runner.invoke(app, ["login"])
+
+        assert result.exit_code == 0
+        assert mock_open.called
+        assert mock_open.call_args.args == ("http://test.com",)
+        assert "Now you are logged in!" in result.output