Merge branch 'main' into ref/update-login

Author: alejsdev

release-notes.md

@@ -4,8 +4,14 @@
 
 ### Features
 
+* ✨ Add fastapi cloud sub-command. PR [#104](https://github.com/fastapilabs/fastapi-cloud-cli/pull/104) by [@buurro](https://github.com/buurro).
+* ✨ Check if token is expired when checking if user is logged in. PR [#105](https://github.com/fastapilabs/fastapi-cloud-cli/pull/105) by [@patrick91](https://github.com/patrick91).
 * ✨ Support the verification skipped status. PR [#99](https://github.com/fastapilabs/fastapi-cloud-cli/pull/99) by [@DoctorJohn](https://github.com/DoctorJohn).
 
+### Fixes
+
+* ♻️  Clean up code archives after uploading. PR [#106](https://github.com/fastapilabs/fastapi-cloud-cli/pull/106) by [@DoctorJohn](https://github.com/DoctorJohn).
+
 ## 0.3.1
 
 ### Fixes

src/fastapi_cloud_cli/cli.py

@@ -13,17 +13,29 @@
 
 app = typer.Typer(rich_markup_mode="rich")
 
+cloud_app = typer.Typer(
+    rich_markup_mode="rich",
+    help="Manage [bold]FastAPI[/bold] Cloud deployments. 🚀",
+)
 
 # TODO: use the app structure
 
 # Additional commands
+
+# fastapi cloud [command]
+cloud_app.command()(deploy)
+cloud_app.command()(login)
+cloud_app.command()(logout)
+cloud_app.command()(whoami)
+cloud_app.command()(unlink)
+
+cloud_app.add_typer(env_app, name="env")
+
+# fastapi [command]
 app.command()(deploy)
 app.command()(login)
-app.command()(logout)
-app.command()(whoami)
-app.command()(unlink)
 
-app.add_typer(env_app, name="env")
+app.add_typer(cloud_app, name="cloud")
 
 
 def main() -> None:

src/fastapi_cloud_cli/commands/deploy.py

@@ -5,7 +5,6 @@
 import tarfile
 import tempfile
 import time
-import uuid
 from enum import Enum
 from itertools import cycle
 from pathlib import Path
@@ -46,19 +45,14 @@ def _should_exclude_entry(path: Path) -> bool:
     return False
 
 
-def archive(path: Path) -> Path:
+def archive(path: Path, tar_path: Path) -> Path:
     logger.debug("Starting archive creation for path: %s", path)
     files = rignore.walk(
         path,
         should_exclude_entry=_should_exclude_entry,
         additional_ignore_paths=[".fastapicloudignore"],
     )
 
-    temp_dir = tempfile.mkdtemp()
-    logger.debug("Created temp directory: %s", temp_dir)
-
-    name = f"fastapi-cloud-deploy-{uuid.uuid4()}"
-    tar_path = Path(temp_dir) / f"{name}.tar"
     logger.debug("Archive will be created at: %s", tar_path)
 
     file_count = 0
@@ -496,7 +490,7 @@ def _waitlist_form(toolkit: RichToolkit) -> None:
 
         with contextlib.suppress(Exception):
             subprocess.run(
-                ["open", "raycast://confetti"],
+                ["open", "raycast://confetti?emojis=🐔⚡"],
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
                 check=False,
@@ -581,16 +575,19 @@ def deploy(
         if not app:
             toolkit.print_line()
             toolkit.print(
-                "If you deleted this app, you can run [bold]fastapi unlink[/] to unlink the local configuration.",
+                "If you deleted this app, you can run [bold]fastapi cloud unlink[/] to unlink the local configuration.",
                 tag="tip",
             )
             raise typer.Exit(1)
 
-        logger.debug("Creating archive for deployment")
-        archive_path = archive(path or Path.cwd())  # noqa: F841
+        with tempfile.TemporaryDirectory() as temp_dir:
+            logger.debug("Creating archive for deployment")
+            archive_path = Path(temp_dir) / "archive.tar"
+            archive(path or Path.cwd(), archive_path)
 
-        with toolkit.progress(title="Creating deployment") as progress:
-            with handle_http_errors(progress):
+            with toolkit.progress(
+                title="Creating deployment"
+            ) as progress, handle_http_errors(progress):
                 logger.debug("Creating deployment for app: %s", app.id)
                 deployment = _create_deployment(app.id)
 
@@ -602,7 +599,7 @@ def deploy(
 
                 _upload_deployment(deployment.id, archive_path)
 
-            progress.log("Deployment uploaded successfully!")
+                progress.log("Deployment uploaded successfully!")
 
         toolkit.print_line()
 

src/fastapi_cloud_cli/utils/auth.py

@@ -1,4 +1,8 @@
+import base64
+import binascii
+import json
 import logging
+import time
 from typing import Optional
 
 from pydantic import BaseModel
@@ -55,7 +59,64 @@ def get_auth_token() -> Optional[str]:
     return auth_data.access_token
 
 
+def is_token_expired(token: str) -> bool:
+    try:
+        parts = token.split(".")
+
+        if len(parts) != 3:
+            logger.debug("Invalid JWT format: expected 3 parts, got %d", len(parts))
+            return True
+
+        payload = parts[1]
+
+        # Add padding if needed (JWT uses base64url encoding without padding)
+        if padding := len(payload) % 4:
+            payload += "=" * (4 - padding)
+
+        payload = payload.replace("-", "+").replace("_", "/")
+        decoded_bytes = base64.b64decode(payload)
+        payload_data = json.loads(decoded_bytes)
+
+        exp = payload_data.get("exp")
+
+        if exp is None:
+            logger.debug("No 'exp' claim found in token")
+
+            return False
+
+        if not isinstance(exp, int):  # pragma: no cover
+            logger.debug("Invalid 'exp' claim: expected int, got %s", type(exp))
+
+            return True
+
+        current_time = time.time()
+
+        is_expired = current_time >= exp
+
+        logger.debug(
+            "Token expiration check: current=%d, exp=%d, expired=%s",
+            current_time,
+            exp,
+            is_expired,
+        )
+
+        return is_expired
+    except (binascii.Error, json.JSONDecodeError) as e:
+        logger.debug("Error parsing JWT token: %s", e)
+
+        return True
+
+
 def is_logged_in() -> bool:
-    result = get_auth_token() is not None
-    logger.debug("Login status: %s", result)
-    return result
+    token = get_auth_token()
+
+    if 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
+
+    logger.debug("Login status: True")
+    return True

tests/conftest.py

@@ -7,6 +7,8 @@
 import pytest
 from typer import rich_utils
 
+from .utils import create_jwt_token
+
 
 @pytest.fixture(autouse=True)
 def reset_syspath() -> Generator[None, None, None]:
@@ -26,7 +28,9 @@ def setup_terminal() -> None:
 
 @pytest.fixture
 def logged_in_cli(temp_auth_config: Path) -> Generator[None, None, None]:
-    temp_auth_config.write_text('{"access_token": "test_token_12345"}')
+    valid_token = create_jwt_token({"sub": "test_user_12345"})
+
+    temp_auth_config.write_text(f'{{"access_token": "{valid_token}"}}')
 
     yield
 

tests/test_archive.py

@@ -1,76 +1,89 @@
 import tarfile
 from pathlib import Path
 
+import pytest
+
 from fastapi_cloud_cli.commands.deploy import archive
 
 
-def test_archive_creates_tar_file(tmp_path: Path) -> None:
-    (tmp_path / "main.py").write_text("print('hello')")
-    (tmp_path / "config.json").write_text('{"key": "value"}')
-    (tmp_path / "subdir").mkdir()
-    (tmp_path / "subdir" / "utils.py").write_text("def helper(): pass")
+@pytest.fixture
+def src_path(tmp_path: Path) -> Path:
+    path = tmp_path / "source"
+    path.mkdir()
+    return path
+
+
+@pytest.fixture
+def tar_path(tmp_path: Path) -> Path:
+    return tmp_path / "archive.tar"
+
 
-    tar_path = archive(tmp_path)
+def test_archive_creates_tar_file(src_path: Path, tar_path: Path) -> None:
+    (src_path / "main.py").write_text("print('hello')")
+    (src_path / "config.json").write_text('{"key": "value"}')
+    (src_path / "subdir").mkdir()
+    (src_path / "subdir" / "utils.py").write_text("def helper(): pass")
 
+    archive(src_path, tar_path)
     assert tar_path.exists()
-    assert tar_path.suffix == ".tar"
-    assert tar_path.name.startswith("fastapi-cloud-deploy-")
 
 
-def test_archive_excludes_venv_and_similar_folders(tmp_path: Path) -> None:
+def test_archive_excludes_venv_and_similar_folders(
+    src_path: Path, tar_path: Path
+) -> None:
     """Should exclude .venv directory from archive."""
     # the only files we want to include
-    (tmp_path / "main.py").write_text("print('hello')")
-    (tmp_path / "static").mkdir()
-    (tmp_path / "static" / "index.html").write_text("<html></html>")
+    (src_path / "main.py").write_text("print('hello')")
+    (src_path / "static").mkdir()
+    (src_path / "static" / "index.html").write_text("<html></html>")
     # virtualenv
-    (tmp_path / ".venv").mkdir()
-    (tmp_path / ".venv" / "lib").mkdir()
-    (tmp_path / ".venv" / "lib" / "package.py").write_text("# package")
+    (src_path / ".venv").mkdir()
+    (src_path / ".venv" / "lib").mkdir()
+    (src_path / ".venv" / "lib" / "package.py").write_text("# package")
     # pycache
-    (tmp_path / "__pycache__").mkdir()
-    (tmp_path / "__pycache__" / "main.cpython-311.pyc").write_text("bytecode")
+    (src_path / "__pycache__").mkdir()
+    (src_path / "__pycache__" / "main.cpython-311.pyc").write_text("bytecode")
     # pyc files
-    (tmp_path / "main.pyc").write_text("bytecode")
+    (src_path / "main.pyc").write_text("bytecode")
     # mypy/pytest
-    (tmp_path / ".mypy_cache").mkdir()
-    (tmp_path / ".mypy_cache" / "file.json").write_text("{}")
-    (tmp_path / ".pytest_cache").mkdir()
-    (tmp_path / ".pytest_cache" / "cache.db").write_text("data")
+    (src_path / ".mypy_cache").mkdir()
+    (src_path / ".mypy_cache" / "file.json").write_text("{}")
+    (src_path / ".pytest_cache").mkdir()
+    (src_path / ".pytest_cache" / "cache.db").write_text("data")
 
-    tar_path = archive(tmp_path)
+    archive(src_path, tar_path)
 
     with tarfile.open(tar_path, "r") as tar:
         names = tar.getnames()
         assert set(names) == {"main.py", "static/index.html"}
 
 
-def test_archive_preserves_relative_paths(tmp_path: Path) -> None:
-    (tmp_path / "src").mkdir()
-    (tmp_path / "src" / "app").mkdir()
-    (tmp_path / "src" / "app" / "main.py").write_text("print('hello')")
+def test_archive_preserves_relative_paths(src_path: Path, tar_path: Path) -> None:
+    (src_path / "src").mkdir()
+    (src_path / "src" / "app").mkdir()
+    (src_path / "src" / "app" / "main.py").write_text("print('hello')")
 
-    tar_path = archive(tmp_path)
+    archive(src_path, tar_path)
 
     with tarfile.open(tar_path, "r") as tar:
         names = tar.getnames()
         assert names == ["src/app/main.py"]
 
 
-def test_archive_respects_fastapicloudignore(tmp_path: Path) -> None:
+def test_archive_respects_fastapicloudignore(src_path: Path, tar_path: Path) -> None:
     """Should exclude files specified in .fastapicloudignore."""
     # Create test files
-    (tmp_path / "main.py").write_text("print('hello')")
-    (tmp_path / "config.py").write_text("CONFIG = 'value'")
-    (tmp_path / "secrets.env").write_text("SECRET_KEY=xyz")
-    (tmp_path / "data").mkdir()
-    (tmp_path / "data" / "file.txt").write_text("data")
+    (src_path / "main.py").write_text("print('hello')")
+    (src_path / "config.py").write_text("CONFIG = 'value'")
+    (src_path / "secrets.env").write_text("SECRET_KEY=xyz")
+    (src_path / "data").mkdir()
+    (src_path / "data" / "file.txt").write_text("data")
 
     # Create .fastapicloudignore file
-    (tmp_path / ".fastapicloudignore").write_text("secrets.env\ndata/\n")
+    (src_path / ".fastapicloudignore").write_text("secrets.env\ndata/\n")
 
     # Create archive
-    tar_path = archive(tmp_path)
+    archive(src_path, tar_path)
 
     # Verify ignored files are excluded
     with tarfile.open(tar_path, "r") as tar:
@@ -81,21 +94,23 @@ def test_archive_respects_fastapicloudignore(tmp_path: Path) -> None:
         }
 
 
-def test_archive_respects_fastapicloudignore_unignore(tmp_path: Path) -> None:
+def test_archive_respects_fastapicloudignore_unignore(
+    src_path: Path, tar_path: Path
+) -> None:
     """Test we can use .fastapicloudignore to unignore files inside .gitignore"""
     # Create test files
-    (tmp_path / "main.py").write_text("print('hello')")
-    (tmp_path / "static/build").mkdir(exist_ok=True, parents=True)
-    (tmp_path / "static/build/style.css").write_text("body { background: #bada55 }")
+    (src_path / "main.py").write_text("print('hello')")
+    (src_path / "static/build").mkdir(exist_ok=True, parents=True)
+    (src_path / "static/build/style.css").write_text("body { background: #bada55 }")
     # Rignore needs a .git folder to make .gitignore work
-    (tmp_path / ".git").mkdir(exist_ok=True, parents=True)
-    (tmp_path / ".gitignore").write_text("build/")
+    (src_path / ".git").mkdir(exist_ok=True, parents=True)
+    (src_path / ".gitignore").write_text("build/")
 
     # Create .fastapicloudignore file
-    (tmp_path / ".fastapicloudignore").write_text("!static/build")
+    (src_path / ".fastapicloudignore").write_text("!static/build")
 
     # Create archive
-    tar_path = archive(tmp_path)
+    archive(src_path, tar_path)
 
     # Verify ignored files are excluded
     with tarfile.open(tar_path, "r") as tar: