Enable branch deploy functionality.

cased/commands/deploy.py

@@ -1,18 +1,24 @@
+import sys
+
 import click
 import questionary
 from rich.console import Console
 
 from cased.utils.api import CasedAPI
 from cased.utils.auth import validate_credentials
+from cased.utils.constants import CasedConstants
 from cased.utils.progress import run_process_with_status_bar
 
 console = Console()
 
 
-def _build_questionary_choices():
+def _build_questionary_choices(project):
     api_client = CasedAPI()
     data = run_process_with_status_bar(
-        api_client.get_branches, "Fetching branches...", timeout=10
+        api_client.get_branches,
+        f"Fetching branches for project {project}...",
+        timeout=10,
+        project_name=project,
     )
     branches = data.get("pull_requests", [])
     deployable_branches = [
@@ -25,8 +31,10 @@ def _build_questionary_choices():
     ]
 
     if not choices:
-        console.print("[red]No deployable branches available.[/red]")
-        return
+        console.print(
+            f"[red]No branches available for project {project}. Please see more details at {CasedConstants.BASE_URL}/deployments/{project} [/red]"  # noqa: E501
+        )
+        sys.exit(1)
 
     selected = questionary.select("Select a branch to deploy:", choices=choices).ask()
 
@@ -52,10 +60,11 @@ def _build_questionary_choices():
 
 
 @click.command()
+@click.option("--project", help="Project name to deploy")
 @click.option("--branch", help="Branch to deploy")
 @click.option("--target", help="Target environment for deployment")
-@validate_credentials(check_project_set=False)
-def deploy(branch, target):
+@validate_credentials(check_project_set=True)
+def deploy(project, branch, target):
     """
     Deploy a branch to a target environment.
 
@@ -70,14 +79,14 @@ def deploy(branch, target):
         cased deploy --branch feature-branch-1 --target dev
     """  # noqa: E501
     if not branch and not target:
-        branch, target = _build_questionary_choices()
+        branch, target = _build_questionary_choices(project)
 
     console.print(
         f"Preparing to deploy [cyan]{branch}[/cyan] to [yellow]{target}[/yellow]"
     )
 
     if branch and target:
-        CasedAPI().deploy_branch(branch, target)
+        CasedAPI().deploy_branch(project, branch, target)
         console.print("[green]Dispatch succeeded. Starting deployment...[/green]")
     else:
         console.print("[red]Deployment dispatch failed. Please try again later.[/red]")

cased/commands/resources.py

@@ -89,6 +89,8 @@ def projects(details=True):
     This command shows a list of available projects and allows you to select one as your current working project.
     If a project is already selected, it will be highlighted in the list.
     The selected project's name and ID are stored as environment variables for future use.
+
+    Use the --details or --d option to show detailed information about each project, by default it is set to True.
     """  # noqa: E501
     # Check if a project is already selected
     config = load_config()
@@ -195,23 +197,8 @@ def deployments(limit, project, target):
     such as begin time, end time, deployer, status, branch, and target.
 
     Use the --limit option to specify the number of deployments to display.
+    Use the --project and --target options to filter deployments by project and target.
     """
-    if not target:
-        targets = CasedAPI().get_targets(project).get("targets", [])
-        if not targets:
-            console.print(
-                f"[yellow]No targets available. You can add a target by going to {CasedConstants.BASE_URL}deployments/{project}/targets/new[/yellow]"  # noqa: E501
-            )
-            return
-
-        selection = questionary.select(
-            "Select a project:", choices=targets, style=custom_style
-        ).ask()
-        if not selection:
-            console.print("[yellow]No target selected.[/yellow]")
-            return
-        target = selection
-
     data = (
         CasedAPI()
         .get_deployments(project_name=project, target_name=target)
@@ -318,55 +305,39 @@ def targets(project):
 @click.command()
 @click.option("--limit", default=5, help="Number of branches to show")
 @click.option("--project", default="", help="Project name to filter branches")
-@click.option("--target", default="", help="Target name to filter branches")
 @validate_credentials(check_project_set=True)
-def branches(limit, project, target):
+def branches(limit, project):
     """
     Display active branches.
 
     This command shows a table of active branches, including information
     such as name, author, PR number, PR title, deployable status, and various checks.
 
     Use the --limit option to specify the number of branches to display.
+    Use the --project option to filter branches by project.
     """
-    if not target:
-        targets = CasedAPI().get_targets(project).get("targets", [])
-        if not targets:
-            console.print(
-                f"[yellow]No targets available. You can add a target by going to {CasedConstants.BASE_URL+'deployments/'+project+'/targets/new'}[/yellow]"  # noqa: E501
-            )
-            return
-
-        selection = questionary.select(
-            "Select a project:", choices=targets, style=custom_style
-        ).ask()
-        if not selection:
-            console.print("[yellow]No target selected.[/yellow]")
-            return
-        target = selection
-
     data = run_process_with_status_bar(
         CasedAPI().get_branches,
         "Fetching branches...",
         timeout=10,
         project_name=project,
-        target_name=target,
     )
     branches = data.get("pull_requests", [])
+    print(len(branches))
+
+    table = Table(title="Active Branches")
+
+    table.add_column("Name", style="cyan")
+    table.add_column("Author", style="magenta")
+    table.add_column("PR Number", style="yellow")
+    table.add_column("PR Title", style="green")
+    table.add_column("Deployable", style="blue")
+    table.add_column("Mergeable", style="blue")
+    table.add_column("Checks", style="cyan")
     for idx, branch in enumerate(branches):
         if idx == limit:
             break
 
-        table = Table(title="Active Branches")
-
-        table.add_column("Name", style="cyan")
-        table.add_column("Author", style="magenta")
-        table.add_column("PR Number", style="yellow")
-        table.add_column("PR Title", style="green")
-        table.add_column("Deployable", style="blue")
-        table.add_column("Mergeable", style="blue")
-        table.add_column("Checks", style="cyan")
-
         table.add_row(
             branch.get("branch_name"),
             branch.get("owner"),

cased/utils/api.py

@@ -1,11 +1,9 @@
-import sys
-
-import click
 import requests
 from rich.console import Console
 
 from cased.utils.config import load_config
 from cased.utils.constants import CasedConstants
+from cased.utils.exception import CasedAPIError
 
 console = Console()
 
@@ -28,66 +26,63 @@ def __init__(self):
             "Accept": "application/json",
         }
 
-    def get_branches(self, project_name, target_name):
-        query_params = {"project_name": project_name, "target_name": target_name}
-        response = requests.get(
-            f"{CasedConstants.API_BASE_URL}/branches",
-            headers=self.request_headers,
-            params=query_params,
-        )
-        if response.status_code == 200:
+    def _make_request(self, resource_name, method, url, **kwargs):
+        response = requests.request(method, url, headers=self.request_headers, **kwargs)
+        if response.status_code in [200, 201]:
             return response.json()
         else:
-            click.echo("Failed to fetch branches. Please try again.")
-            sys.exit(1)
+            raise CasedAPIError(
+                f"Failed to fetch {resource_name} from {url}",
+                response.status_code,
+                response.json(),
+            )
+
+    def get_branches(self, project_name):
+        query_params = {"project_name": project_name}
+        return self._make_request(
+            resource_name="branches",
+            method="GET",
+            url=f"{CasedConstants.API_BASE_URL}/branches",
+            params=query_params,
+        )
 
     def get_projects(self):
-        response = requests.get(
-            f"{CasedConstants.BASE_URL}/deployments", headers=self.request_headers
+        return self._make_request(
+            resource_name="projects",
+            method="GET",
+            url=f"{CasedConstants.BASE_URL}/deployments",
         )
-        if response.status_code == 200:
-            return response.json()
-        else:
-            click.echo("Failed to fetch projects. Please try again.")
-            sys.exit(1)
 
     def get_targets(self, project_name):
         params = {"project_name": project_name}
-        response = requests.get(
-            f"{CasedConstants.API_BASE_URL}/targets",
-            headers=self.request_headers,
+        return self._make_request(
+            resource_name="targets",
+            method="GET",
+            url=f"{CasedConstants.API_BASE_URL}/targets",
             params=params,
         )
-        if response.status_code == 200:
-            return response.json()
-        else:
-            click.echo("Failed to fetch targets. Please try again.")
-            sys.exit(1)
 
-    def get_deployments(self, project_name, target_name):
-        response = requests.get(
-            f"{CasedConstants.API_BASE_URL}/targets/{project_name}/{target_name}/deployments/",
-            headers=self.request_headers,
+    def get_deployments(self, project_name, target_name=None):
+        params = {"project_name": project_name, "target_name": target_name}
+        return self._make_request(
+            resource_name="deployments",
+            method="GET",
+            url=f"{CasedConstants.API_BASE_URL}/deployments",
+            params=params,
         )
-        if response.status_code == 200:
-            return response.json()
-        else:
-            click.echo("Failed to fetch deployments. Please try again.")
-            sys.exit(1)
 
-    def deploy_branch(self, branch_name, target_name):
-        # Implement branch deployment logic here
-        response = requests.post(
-            f"{CasedConstants.API_BASE_URL}/branch-deploys/",
-            json={"branch_name": branch_name, "target_name": target_name},
-            headers=self.request_headers,
+    def deploy_branch(self, project_name, branch_name, target_name):
+        json = {
+            "project_name": project_name,
+            "branch_name": branch_name,
+            "target_name": target_name,
+        }
+        return self._make_request(
+            resource_name="branch_deploy",
+            method="POST",
+            url=f"{CasedConstants.API_BASE_URL}/branch-deploys",
+            json=json,
         )
-        if response.status_code == 200:
-            click.echo(
-                f"Successfully deployed branch '{branch_name}' to target '{target_name}'!"
-            )
-        else:
-            sys.exit(1)
 
     def create_secrets(self, project_name: str, secrets: list):
         payload = {

cased/utils/exception.py

@@ -0,0 +1,30 @@
+from typing import Any, Optional
+
+
+class CasedAPIError(Exception):
+    def __init__(
+        self, message: str, status_code: Optional[int] = None, response_body: Any = None
+    ):
+        """
+        Initialize the CasedAPIError.
+
+        Args:
+            message (str): The error message.
+            status_code (Optional[int]): The HTTP status code of the failed request.
+            response_body (Any): The response body from the failed request.
+        """
+        self.message = message
+        self.status_code = status_code
+        self.response_body = response_body
+        super().__init__(self.message)
+
+    # TODO: make this specific based on status codes returned from our API,
+    # right now it is too generic.
+    def __str__(self):
+        """Return a string representation of the error."""
+        error_msg = self.message
+        if self.status_code:
+            error_msg += f" (Status code: {self.status_code})"
+        if self.response_body:
+            error_msg += f"\nResponse body: {self.response_body}"
+        return error_msg