♻️ Refactor authentication

Author: patrick91

src/fastapi_cloud_cli/commands/deploy.py

@@ -21,7 +21,7 @@
 from fastapi_cloud_cli.commands.login import login
 from fastapi_cloud_cli.utils.api import APIClient, BuildLogError, TooManyRetriesError
 from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config, write_app_config
-from fastapi_cloud_cli.utils.auth import is_logged_in
+from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 
 logger = logging.getLogger(__name__)
@@ -534,8 +534,10 @@ def deploy(
     logger.debug("Deploy command started")
     logger.debug("Deploy path: %s, skip_wait: %s", path, skip_wait)
 
+    identity = Identity()
+
     with get_rich_toolkit() as toolkit:
-        if not is_logged_in():
+        if not identity.is_logged_in():
             logger.debug("User not logged in, prompting for login or waitlist")
 
             toolkit.print_title("Welcome to FastAPI Cloud!", tag="FastAPI")

src/fastapi_cloud_cli/commands/env.py

@@ -7,7 +7,7 @@
 
 from fastapi_cloud_cli.utils.api import APIClient
 from fastapi_cloud_cli.utils.apps import get_app_config
-from fastapi_cloud_cli.utils.auth import is_logged_in
+from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 from fastapi_cloud_cli.utils.env import validate_environment_variable_name
 
@@ -70,8 +70,10 @@ def list(
     List the environment variables for the app.
     """
 
+    identity = Identity()
+
     with get_rich_toolkit(minimal=True) as toolkit:
-        if not is_logged_in():
+        if not identity.is_logged_in():
             toolkit.print(
                 "No credentials found. Use [blue]`fastapi login`[/] to login.",
                 tag="auth",
@@ -123,9 +125,10 @@ def delete(
     Delete an environment variable from the app.
     """
 
+    identity = Identity()
+
     with get_rich_toolkit(minimal=True) as toolkit:
-        # TODO: maybe this logic can be extracted to a function
-        if not is_logged_in():
+        if not identity.is_logged_in():
             toolkit.print(
                 "No credentials found. Use [blue]`fastapi login`[/] to login.",
                 tag="auth",
@@ -208,8 +211,10 @@ def set(
     Set an environment variable for the app.
     """
 
+    identity = Identity()
+
     with get_rich_toolkit(minimal=True) as toolkit:
-        if not is_logged_in():
+        if not identity.is_logged_in():
             toolkit.print(
                 "No credentials found. Use [blue]`fastapi login`[/] to login.",
                 tag="auth",

src/fastapi_cloud_cli/commands/login.py

@@ -8,13 +8,7 @@
 
 from fastapi_cloud_cli.config import Settings
 from fastapi_cloud_cli.utils.api import APIClient
-from fastapi_cloud_cli.utils.auth import (
-    AuthConfig,
-    get_auth_token,
-    is_logged_in,
-    is_token_expired,
-    write_auth_config,
-)
+from fastapi_cloud_cli.utils.auth import AuthConfig, Identity, write_auth_config
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 
 logger = logging.getLogger(__name__)
@@ -82,13 +76,14 @@ def login() -> Any:
     """
     Login to FastAPI Cloud. 🚀
     """
-    token = get_auth_token()
-    if token is not None and is_token_expired(token):
+    identity = Identity()
+
+    if identity.is_expired():
         with get_rich_toolkit(minimal=True) as toolkit:
             toolkit.print("Your session has expired. Logging in again...")
             toolkit.print_line()
 
-    if is_logged_in():
+    if identity.is_logged_in():
         with get_rich_toolkit(minimal=True) as toolkit:
             toolkit.print("You are already logged in.")
             toolkit.print(

src/fastapi_cloud_cli/commands/whoami.py

@@ -5,14 +5,16 @@
 from rich_toolkit.progress import Progress
 
 from fastapi_cloud_cli.utils.api import APIClient
-from fastapi_cloud_cli.utils.auth import is_logged_in
+from fastapi_cloud_cli.utils.auth import Identity
 from fastapi_cloud_cli.utils.cli import handle_http_errors
 
 logger = logging.getLogger(__name__)
 
 
 def whoami() -> Any:
-    if not is_logged_in():
+    identity = Identity()
+
+    if not identity.is_logged_in():
         print("No credentials found. Use [blue]`fastapi login`[/] to login.")
         return
 

src/fastapi_cloud_cli/utils/api.py

@@ -20,7 +20,8 @@
 
 from fastapi_cloud_cli import __version__
 from fastapi_cloud_cli.config import Settings
-from fastapi_cloud_cli.utils.auth import get_auth_token
+
+from .auth import Identity
 
 logger = logging.getLogger(__name__)
 
@@ -132,14 +133,13 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> Generator[T, None, None]:
 class APIClient(httpx.Client):
     def __init__(self) -> None:
         settings = Settings.get()
-
-        token = get_auth_token()
+        identity = Identity()
 
         super().__init__(
             base_url=settings.base_api_url,
             timeout=httpx.Timeout(20),
             headers={
-                "Authorization": f"Bearer {token}",
+                "Authorization": f"Bearer {identity.token}",
                 "User-Agent": f"fastapi-cloud-cli/{__version__}",
             },
         )

src/fastapi_cloud_cli/utils/auth.py

@@ -47,7 +47,7 @@ def read_auth_config() -> Optional[AuthConfig]:
     return AuthConfig.model_validate_json(auth_path.read_text(encoding="utf-8"))
 
 
-def get_auth_token() -> Optional[str]:
+def _get_auth_token() -> Optional[str]:
     logger.debug("Getting auth token")
     auth_data = read_auth_config()
 
@@ -59,7 +59,7 @@ def get_auth_token() -> Optional[str]:
     return auth_data.access_token
 
 
-def is_token_expired(token: str) -> bool:
+def _is_token_expired(token: str) -> bool:
     try:
         parts = token.split(".")
 
@@ -107,16 +107,24 @@ def is_token_expired(token: str) -> bool:
         return True
 
 
-def is_logged_in() -> bool:
-    token = get_auth_token()
+class Identity:
+    def __init__(self) -> None:
+        self.token = _get_auth_token()
 
-    if token is None:
-        logger.debug("Login status: False (no token)")
-        return False
+    def is_expired(self) -> bool:
+        if not self.token:
+            return True
+
+        return _is_token_expired(self.token)
+
+    def is_logged_in(self) -> bool:
+        if self.token is None:
+            logger.debug("Login status: False (no token)")
+            return False
 
-    if is_token_expired(token):
-        logger.debug("Login status: False (token expired)")
-        return False
+        if self.is_expired():
+            logger.debug("Login status: False (token expired)")
+            return False
 
-    logger.debug("Login status: True")
-    return True
+        logger.debug("Login status: True")
+        return True

src/fastapi_cloud_cli/utils/sentry.py

@@ -1,14 +1,16 @@
 import sentry_sdk
 from sentry_sdk.integrations.typer import TyperIntegration
 
-from .auth import is_logged_in
+from fastapi_cloud_cli.utils.auth import Identity
 
 SENTRY_DSN = "https://230250605ea4b58a0b69c768e9ec1168@o4506985151856640.ingest.us.sentry.io/4508449198899200"
 
 
 def init_sentry() -> None:
     """Initialize Sentry error tracking only if user is logged in."""
-    if not is_logged_in():
+    identity = Identity()
+
+    if not identity.is_logged_in():
         return
 
     sentry_sdk.init(

tests/test_auth.py

@@ -6,8 +6,8 @@
 
 from fastapi_cloud_cli.utils.auth import (
     AuthConfig,
-    is_logged_in,
-    is_token_expired,
+    Identity,
+    _is_token_expired,
     write_auth_config,
 )
 
@@ -19,21 +19,21 @@ def test_is_token_expired_with_valid_token() -> None:
 
     token = create_jwt_token({"exp": future_exp, "sub": "test_user"})
 
-    assert not is_token_expired(token)
+    assert not _is_token_expired(token)
 
 
 def test_is_token_expired_with_expired_token() -> None:
     past_exp = int(time.time()) - 3600
     token = create_jwt_token({"exp": past_exp, "sub": "test_user"})
 
-    assert is_token_expired(token)
+    assert _is_token_expired(token)
 
 
 def test_is_token_expired_with_no_exp_claim() -> None:
     token = create_jwt_token({"sub": "test_user"})
 
     # Tokens without exp claim should be considered valid
-    assert not is_token_expired(token)
+    assert not _is_token_expired(token)
 
 
 @pytest.mark.parametrize(
@@ -47,12 +47,12 @@ def test_is_token_expired_with_no_exp_claim() -> None:
     ],
 )
 def test_is_token_expired_with_malformed_token(token: str) -> None:
-    assert is_token_expired(token)
+    assert _is_token_expired(token)
 
 
 def test_is_token_expired_with_invalid_base64() -> None:
     token = "header.!!!invalid_signature!!!.signature"
-    assert is_token_expired(token)
+    assert _is_token_expired(token)
 
 
 def test_is_token_expired_with_invalid_json() -> None:
@@ -61,12 +61,26 @@ def test_is_token_expired_with_invalid_json() -> None:
     signature = base64.urlsafe_b64encode(b"signature").decode().rstrip("=")
     token = f"{header_encoded}.{payload_encoded}.{signature}"
 
-    assert is_token_expired(token)
+    assert _is_token_expired(token)
+
+
+def test_is_token_expired_edge_case_exact_expiration() -> None:
+    current_time = int(time.time())
+    token = create_jwt_token({"exp": current_time, "sub": "test_user"})
+
+    assert _is_token_expired(token)
+
+
+def test_is_token_expired_edge_case_one_second_before() -> None:
+    current_time = int(time.time())
+    token = create_jwt_token({"exp": current_time + 1, "sub": "test_user"})
+
+    assert not _is_token_expired(token)
 
 
 def test_is_logged_in_with_no_token(temp_auth_config: Path) -> None:
     assert not temp_auth_config.exists()
-    assert not is_logged_in()
+    assert not Identity().is_logged_in()
 
 
 def test_is_logged_in_with_valid_token(temp_auth_config: Path) -> None:
@@ -75,7 +89,7 @@ def test_is_logged_in_with_valid_token(temp_auth_config: Path) -> None:
 
     write_auth_config(AuthConfig(access_token=token))
 
-    assert is_logged_in()
+    assert Identity().is_logged_in()
 
 
 def test_is_logged_in_with_expired_token(temp_auth_config: Path) -> None:
@@ -84,24 +98,10 @@ def test_is_logged_in_with_expired_token(temp_auth_config: Path) -> None:
 
     write_auth_config(AuthConfig(access_token=token))
 
-    assert not is_logged_in()
+    assert not Identity().is_logged_in()
 
 
 def test_is_logged_in_with_malformed_token(temp_auth_config: Path) -> None:
     write_auth_config(AuthConfig(access_token="not.a.valid.token"))
 
-    assert not is_logged_in()
-
-
-def test_is_token_expired_edge_case_exact_expiration() -> None:
-    current_time = int(time.time())
-    token = create_jwt_token({"exp": current_time, "sub": "test_user"})
-
-    assert is_token_expired(token)
-
-
-def test_is_token_expired_edge_case_one_second_before() -> None:
-    current_time = int(time.time())
-    token = create_jwt_token({"exp": current_time + 1, "sub": "test_user"})
-
-    assert not is_token_expired(token)
+    assert not Identity().is_logged_in()