feat: flash app build uploads and deployments with state manager

Author: jhcipar

src/tetra_rp/cli/commands/build.py

@@ -5,6 +5,7 @@
 import subprocess
 import sys
 import tarfile
+import asyncio
 from pathlib import Path
 
 import typer
@@ -14,6 +15,8 @@
 from rich.table import Table
 
 from ..utils.ignore import get_file_tree, load_ignore_patterns
+from ..utils.app import discover_flash_project
+from tetra_rp.core.resources.app import FlashApp
 
 console = Console()
 
@@ -31,6 +34,9 @@ def build_command(
     output_name: str | None = typer.Option(
         None, "--output", "-o", help="Custom archive name (default: archive.tar.gz)"
     ),
+    local_only: bool = typer.Option(
+        False, "--local-only", help="Only store build tarball locally and dont upload to Runpod"
+    ),
 ):
     """
     Build Flash application for deployment.
@@ -150,6 +156,11 @@ def build_command(
 
         # Success summary
         _display_build_summary(archive_path, app_name, len(files), len(requirements))
+        
+        if local_only:
+            return
+
+        asyncio.run(upload_build(app_name, archive_path))
 
     except KeyboardInterrupt:
         console.print("\n[yellow]Build cancelled by user[/yellow]")
@@ -161,21 +172,9 @@ def build_command(
         console.print(traceback.format_exc())
         raise typer.Exit(1)
 
-
-def discover_flash_project() -> tuple[Path, str]:
-    """
-    Discover Flash project directory and app name.
-
-    Returns:
-        Tuple of (project_dir, app_name)
-
-    Raises:
-        typer.Exit: If not in a Flash project directory
-    """
-    project_dir = Path.cwd()
-    app_name = project_dir.name
-
-    return project_dir, app_name
+async def upload_build(app_name: str, build_path: str | Path):
+    app = await FlashApp.from_name(app_name) 
+    await app.upload_build(build_path)
 
 
 def validate_project_structure(project_dir: Path) -> bool:

src/tetra_rp/cli/commands/deploy.py

@@ -4,8 +4,12 @@
 from rich.console import Console
 from rich.table import Table
 from rich.panel import Panel
+from rich.progress import Progress, SpinnerColumn, TextColumn
 import questionary
+import asyncio
+from typing import Optional
 
+from ..utils.app import discover_flash_project
 from ..utils.deployment import (
     get_deployment_environments,
     create_deployment_environment,
@@ -15,8 +19,71 @@
     get_environment_info,
 )
 
+from tetra_rp.core.resources import FlashApp
+
 console = Console()
 
+deploy_app = typer.Typer(short_help="Deploy flash application code to managed environments on Runpod")
+
+@deploy_app.callback(invoke_without_command=True)
+def deploy_callback(
+        ctx: typer.Context,
+        app_name: str = typer.Option(None, "--app-name", "-a", help="Flash app name to deploy a build for"),
+        env_name: str = typer.Option("dev", "--env-name", "-e", help="Flash env name to deploy a build to. If not provided, default to 'dev'"),
+        build_id: str = typer.Option(None, "--build-id", "-b", help="Flash build id to deploy to env. If not provided, default to most recently uploaded build"),
+):
+    if ctx.invoked_subcommand is None:
+        deploy_build_sync(app_name, env_name, build_id)
+
+def deploy_build_sync(app_name: str, env_name: str, build_id: str):
+    """
+    Convenience wrapper to run async methods from cli
+    """
+    asyncio.run(deploy_build(app_name, env_name, build_id))
+
+async def deploy_build(app_name: Optional[str] = "", env_name: Optional[str] = "dev", build_id: Optional[str] = None):
+    target_env = env_name or "dev"
+    progress_columns = [SpinnerColumn(), TextColumn("[progress.description]{task.description}")]
+
+    with Progress(*progress_columns, console=console) as progress:
+        task = progress.add_task("Preparing deployment", start=True)
+
+        if not app_name:
+            progress.update(task, description="Discovering Flash project...")
+            _, app_name = discover_flash_project()
+            progress.update(task, description=f"Detected Flash app '{app_name}'")
+
+        progress.update(task, description=f"Fetching Flash app '{app_name}'...")
+        app = await FlashApp.from_name(app_name)
+
+        if not build_id:
+            progress.update(task, description="Fetching available builds...")
+            builds = await app.list_builds()
+            if not builds:
+                progress.update(task, description="No builds found for app")
+                progress.stop_task(task)
+                raise ValueError("No builds found for app. Run 'flash build' first.")
+            build_id = builds[0]["id"]
+            progress.update(task, description=f"Using latest build {build_id}")
+
+        progress.update(
+            task,
+            description=f"Promoting build {build_id} to environment '{target_env}'...",
+        )
+        result = await app.deploy_build_to_environment(build_id, environment_name=target_env)
+        progress.update(
+            task,
+            description=f"[green]✓ Promoted build {build_id} to '{target_env}'",
+        )
+        progress.stop_task(task)
+
+    panel_content = (
+        f"Promoted build [bold]{build_id}[/bold] to environment [bold]{target_env}[/bold]\n"
+        f"Flash app: [bold]{app_name}[/bold]"
+    )
+    console.print(Panel(panel_content, title="Deployment Promotion", expand=False))
+
+    return result
 
 def list_command():
     """Show available deployment environments."""

src/tetra_rp/cli/commands/init.py

@@ -6,6 +6,8 @@
 from rich.console import Console
 from rich.panel import Panel
 from rich.table import Table
+from tetra_rp.core.resources.app import FlashApp
+import asyncio
 
 from ..utils.skeleton import create_project_skeleton
 from ..utils.conda import (
@@ -27,6 +29,9 @@
     "aiohttp>=3.9.0",
 ]
 
+async def init_app(app_name: str):
+    app = await FlashApp.create(app_name)
+    await app.create_environment("dev")
 
 def init_command(
     project_name: str = typer.Argument(..., help="Project name"),
@@ -36,6 +41,9 @@ def init_command(
     force: bool = typer.Option(
         False, "--force", "-f", help="Overwrite existing directory"
     ),
+    local_only: bool = typer.Option(
+        False, "--local-only", help="Do not create remote app on Runpod"
+    ),
 ):
     """Create new Flash project with Flash Server and GPU workers."""
 
@@ -95,6 +103,9 @@ def init_command(
                             "You can manually install dependencies: pip install -r requirements.txt"
                         )
 
+    if not local_only:
+        asyncio.run(init_app(project_name))
+
     # Success output
     panel_content = (
         f"Flash project '[bold]{project_name}[/bold]' created successfully!\n\n"

src/tetra_rp/cli/commands/state.py

@@ -1,30 +1,22 @@
 """Flash deploy command - Deploy Flash project to production."""
 import asyncio
-import json
-from pathlib import Path
-from typing import Optional
 
 from tetra_rp import FlashApp
 
 import typer
 from rich.console import Console
-from rich.panel import Panel
-from rich.progress import Progress, SpinnerColumn, TextColumn
 from rich.table import Table
 
 console = Console()
-ls_app = typer.Typer()
 
-# flash ls 
-# flash ls envs --app-name
-# flash ls builds --app-name
+ls_app = typer.Typer(short_help="List existing apps, environments, and builds")
 
 @ls_app.callback(invoke_without_command=True)
 def list_callback(ctx: typer.Context):
     if ctx.invoked_subcommand is None:
         asyncio.run(list_flash_apps())
 
-@ls_app.command("ls")
+@ls_app.command()
 def ls():
     return asyncio.run(list_flash_apps())
 

src/tetra_rp/cli/main.py

@@ -11,6 +11,7 @@
     build,
     resource,
     deploy,
+    state,
 )
 
 
@@ -39,22 +40,8 @@ def get_version() -> str:
 app.command("report")(resource.report_command)
 app.command("clean")(resource.clean_command)
 
-# command: flash deploy
-deploy_app = typer.Typer(
-    name="deploy",
-    help="Deployment environment management commands",
-    no_args_is_help=True,
-)
-
-# command: flash deploy *
-deploy_app.command("list")(deploy.list_command)
-deploy_app.command("new")(deploy.new_command)
-deploy_app.command("send")(deploy.send_command)
-deploy_app.command("report")(deploy.report_command)
-deploy_app.command("rollback")(deploy.rollback_command)
-deploy_app.command("remove")(deploy.remove_command)
-
-app.add_typer(deploy_app, name="deploy")
+app.add_typer(deploy.deploy_app, name="deploy")
+app.add_typer(state.ls_app, name="ls")
 
 
 @app.callback(invoke_without_command=True)

src/tetra_rp/cli/utils/app.py

@@ -0,0 +1,18 @@
+from pathlib import Path
+
+def discover_flash_project() -> tuple[Path, str]:
+    """
+    Discover Flash project directory and app name.
+
+    Returns:
+        Tuple of (project_dir, app_name)
+
+    Raises:
+        typer.Exit: If not in a Flash project directory
+    """
+    project_dir = Path.cwd()
+    app_name = project_dir.name
+
+    return project_dir, app_name
+
+

src/tetra_rp/core/api/runpod.py

@@ -6,7 +6,7 @@
 import json
 import logging
 import os
-from typing import Any, Dict, Optional, List
+from typing import Any, Dict, Optional
 
 import aiohttp
 
@@ -356,13 +356,14 @@ async def get_flash_environment_by_name(self, input_data: Dict[str, Any]) -> Dic
                         name
                     }
                 }
+            }
         """
         variables = {"input": input_data}
 
         log.debug(f"Fetching flash environment by name for input: {variables}")
         result = await self._execute_graphql(query, variables)
 
-        return result["flashEnvironment"]
+        return result["flashEnvironmentByName"]
 
     async def get_flash_artifact_url(self, environment_id: str) -> Dict[str, Any]:
         result = await self.get_flash_environment({"flashEnvironmentId": environment_id})
@@ -502,6 +503,21 @@ async def set_environment_state(self, input_data: Dict[str, Any]) -> Dict[str, A
 
         return result["updateFlashEnvironment"]
 
+    async def get_flash_build(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
+        query = """
+        query getFlashBuild($input: String!) {
+                flashBuild(flashBuildId: $input) {
+                    id
+                    name
+            }
+        }
+        """
+        variables = {"input": input_data}
+
+        log.debug(f"Fetching flash build for input: {input_data}")
+        result = await self._execute_graphql(query, variables)
+        return result["flashBuild"]
+
     async def close(self):
         """Close the HTTP session."""
         if self.session and not self.session.closed:
@@ -514,7 +530,6 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.close()
 
 
-
 class RunpodRestClient:
     """
     Runpod REST client for Runpod API.