devbox new subcommand

src/devbox/cli.py

@@ -7,6 +7,7 @@
 import click
 from typing import Optional
 
+from .remote_client import normalize_param_prefix
 from .commands.status import run_status_command
 from .devbox_manager import DevBoxManager
 from .console_output import ConsoleOutput
@@ -15,6 +16,16 @@
 PARAM_PREFIX_ENV_VAR = "DEVBOX_PARAM_PREFIX"
 
 
+def _validate_param_prefix(
+    _ctx: click.Context, _param: click.Parameter, value: str
+) -> str:
+    """Normalize and validate shared ``--param-prefix`` values."""
+    try:
+        return normalize_param_prefix(value)
+    except ValueError as exc:
+        raise click.BadParameter(str(exc)) from exc
+
+
 def param_prefix_option(func):
     """Attach the shared ``--param-prefix`` Click option.
 
@@ -32,6 +43,7 @@ def param_prefix_option(func):
         "--param-prefix",
         default=DEFAULT_PARAM_PREFIX,
         envvar=PARAM_PREFIX_ENV_VAR,
+        callback=_validate_param_prefix,
         show_default=True,
         show_envvar=True,
         help="SSM parameter prefix",
@@ -188,11 +200,20 @@ def launch(
 
 
 @cli.command()
-@click.argument("project")
-@click.option("--base-ami", required=True, help="Base AMI ID for the project")
+@click.argument('project')
+@click.option('--base-ami', required=True, help='Base AMI ID for the project')
+@click.option('--instance-type', help='Default EC2 instance type for future launches')
+@click.option('--key-pair', help='Default SSH key pair name for future launches')
 @param_prefix_option
 @click.pass_context
-def new(ctx, project: str, base_ami: str, param_prefix: str):
+def new(
+    ctx,
+    project: str,
+    base_ami: str,
+    instance_type: Optional[str],
+    key_pair: Optional[str],
+    param_prefix: str,
+):
     """Create a new DevBox project without launching an instance.
 
     PROJECT is the name of the project to create.
@@ -203,7 +224,11 @@ def new(ctx, project: str, base_ami: str, param_prefix: str):
 
     try:
         new_project_programmatic(
-            project=project, base_ami=base_ami, param_prefix=param_prefix
+            project=project,
+            base_ami=base_ami,
+            instance_type=instance_type,
+            key_pair=key_pair,
+            param_prefix=param_prefix
         )
     except Exception as e:
         console.print_error(f"Failed to create project: {str(e)}")

src/devbox/devbox_manager.py

@@ -296,12 +296,9 @@ def delete_ami_and_snapshots(self, ami_id: str) -> Dict[str, Any]:
             raise ValueError("No AMI ID provided.")
 
         try:
-            response = self.ec2.describe_images(ImageIds=[ami_id])
-            images = response.get("Images", [])
-            if not images:
+            image = utils.get_image(ami_id, ec2_client=self.ec2)
+            if image is None:
                 raise utils.ResourceNotFoundError(f"AMI {ami_id} not found.")
-
-            image = images[0]
             snapshot_ids = []
             for mapping in image.get("BlockDeviceMappings", []):
                 ebs = mapping.get("Ebs", {})

src/devbox/launch.py

@@ -167,18 +167,16 @@ def get_volume_info(
         ValueError: If the image is not found or invalid
     """
     try:
-        resp = ec2.describe_images(ImageIds=[image_id])
+        image = utils.get_image(image_id, ec2_client=ec2)
     except ClientError as e:
         raise AWSClientError(
             f"Error fetching image details for {image_id}",
             error_code=e.response.get("Error", {}).get("Code"),
             original_exception=e,
         )
 
-    if not resp.get("Images"):
+    if image is None:
         raise ValueError(f"AMI {image_id} not found")
-
-    image = resp["Images"][0]
     volumes = image.get("BlockDeviceMappings", []).copy()
 
     # Find the largest volume
@@ -860,9 +858,8 @@ def display_instance_info(ec2: Any, instance_id: str, project: str, table: Any)
                         ami_id = resp["Item"].get("AMI", "")
                         if ami_id:
                             try:
-                                ami_resp = ec2.describe_images(ImageIds=[ami_id])
-                                if ami_resp.get("Images"):
-                                    ami = ami_resp["Images"][0]
+                                ami = utils.get_image(ami_id, ec2_client=ec2)
+                                if ami is not None:
                                     ami_name = ami.get("Name", "")
                                     ami_description = ami.get("Description", "")
                                     determined_username = determine_ssh_username(
@@ -880,6 +877,8 @@ def display_instance_info(ec2: Any, instance_id: str, project: str, table: Any)
                                         username = (
                                             "<username>"  # placeholder for unknown
                                         )
+                                else:
+                                    username = "<username>"
                             except Exception:
                                 username = "<username>"
                         else:
@@ -1054,9 +1053,8 @@ def launch_programmatic(
                 existing_username = resp["Item"].get("Username", "")
                 if not existing_username:
                     # Determine username from AMI
-                    ami_resp = aws["ec2"].describe_images(ImageIds=[image_id])
-                    if ami_resp.get("Images"):
-                        ami = ami_resp["Images"][0]
+                    ami = utils.get_image(image_id, ec2_client=aws["ec2"])
+                    if ami is not None:
                         ami_name = ami.get("Name", "")
                         ami_description = ami.get("Description", "")
                         determined_username = determine_ssh_username(
@@ -1069,6 +1067,10 @@ def launch_programmatic(
                                 UpdateExpression="SET Username = :u",
                                 ExpressionAttributeValues={":u": determined_username},
                             )
+                    else:
+                        print(
+                            f"Warning: Could not determine SSH username: AMI {image_id} not found"
+                        )
         except Exception as e:
             print(f"Warning: Could not determine SSH username: {e}")
 

src/devbox/lifecycle/snapshots.py

@@ -9,7 +9,7 @@
 from boto3.dynamodb.conditions import Attr, Key
 from botocore.exceptions import ClientError
 
-from devbox.utils import get_project_tag
+from devbox.utils import get_image, get_project_tag
 
 
 logger = logging.getLogger(__name__)
@@ -177,23 +177,12 @@ def cleanup_ami_and_snapshots(
     logger.info("waiting for ami to vanish ami_id=%s", ami_id)
     for _ in range(config.cleanup_max_attempts):
         time.sleep(config.cleanup_wait_seconds)
-        try:
-            resp = ec2_client.describe_images(ImageIds=[ami_id])
-        except ClientError as exc:
-            code = exc.response.get("Error", {}).get("Code", "")
-            if code.startswith("InvalidAMIID"):
-                logger.info(
-                    "ami no longer exists ami_id=%s error_code=%s", ami_id, code
-                )
-                break
-            raise
-        images = resp.get("Images", [])
-
-        if not images:
+        image = get_image(ami_id, ec2_client=ec2_client)
+        if image is None:
             logger.info("ami no longer exists ami_id=%s", ami_id)
             break
         logger.info(
-            "ami still present ami_id=%s image_count=%s", ami_id, len(images)
+            "ami still present ami_id=%s image_count=%s", ami_id, 1
         )
     else:
         raise RuntimeError(f"Timed out waiting for AMI '{ami_id}' to deregister")
@@ -306,19 +295,12 @@ def make_mapping(item: Dict[str, Any]) -> Dict[str, Any]:
 
     old_ami = main_item.get("AMI")
     if old_ami:
-        desc = []
-        old_ami_missing = False
-        try:
-            desc = ec2_client.describe_images(ImageIds=[old_ami])["Images"]
-        except ClientError as exc:
-            code = exc.response.get("Error", {}).get("Code", "")
-            if code.startswith("InvalidAMIID"):
-                logger.warning("old ami not found ami_id=%s", old_ami)
-                old_ami_missing = True
-            else:
-                raise
-        if desc:
-            image = desc[0]
+        image = get_image(old_ami, ec2_client=ec2_client)
+        old_ami_missing = image is None
+        if old_ami_missing:
+            logger.warning("old ami not found ami_id=%s", old_ami)
+
+        if image is not None:
             if not virtualization_type:
                 virtualization_type = image.get("VirtualizationType")
             if not architecture:
@@ -338,8 +320,6 @@ def make_mapping(item: Dict[str, Any]) -> Dict[str, Any]:
                     old_ami,
                     project,
                 )
-        elif not old_ami_missing:
-            logger.warning("old ami not found ami_id=%s", old_ami)
 
     register_image_args: Dict[str, Any] = {
         "Name": f"{project}-ami",