feat(snapshot): accept instance name and auto-detect volumes

CHANGELOG.md

@@ -6,6 +6,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.6.0] - 2026-04-12
+
+### Added
+- **Snapshot from instance**: `snapshot create` now accepts an instance name to auto-detect and snapshot all attached volumes, with optional `--device` filter to target a specific volume
+- **Partial failure handling**: Multi-volume snapshot creation continues on per-volume errors and reports a summary of successes and failures
+
+### Changed
+- `snapshot create` now accepts instance name as a positional argument (alternative to `--volume-id`)
+- `get_volume_ids` refactored to delegate to new `get_volumes_for_instance` utility
+
 ## [1.4.0] - 2026-01-26
 
 ### Added

CLAUDE.md

@@ -77,7 +77,7 @@ instance_name: str | None = typer.Argument(None, help="Instance name")
 ```
 Then resolve with: `resolve_instance_or_exit(instance_name)` which falls back to the configured default instance.
 
-**Used in**: `instance.py` (status, start, stop, connect, exec, type, terminate), `ami.py` (create), `snapshot.py` (list), `volume.py` (list)
+**Used in**: `instance.py` (status, start, stop, connect, exec, type, terminate), `ami.py` (create), `snapshot.py` (create, list), `volume.py` (list)
 
 #### 2. Required Arguments
 For commands where a value must always be provided:
@@ -89,13 +89,12 @@ The `...` makes the argument required with no default.
 **Used in**: `ami.py` (template-versions, template-info)
 
 #### 3. Required Options
-For commands needing multiple required values that aren't positional:
+For commands needing required values that aren't positional:
 ```python
-volume_id: str = typer.Option(..., "--volume-id", "-v", help="Volume ID (required)")
 name: str = typer.Option(..., "--name", "-n", help="Snapshot name (required)")
 ```
 
-**Used in**: `snapshot.py` (create)
+**Used in**: `snapshot.py` (create: `--name`)
 
 #### 4. Optional Arguments with Interactive Prompts
 For commands where selection from available resources is needed:

README.md

@@ -281,10 +281,22 @@ Resize an EBS volume:
 remote volume resize my-instance --size 100
 ```
 
-Create a snapshot:
+Create snapshots for all volumes attached to an instance:
 
 ```bash
-remote snapshot create vol-12345678
+remote snapshot create my-instance -n my-snapshot
+```
+
+Snapshot a specific device:
+
+```bash
+remote snapshot create my-instance --device /dev/sdf -n my-snapshot
+```
+
+Snapshot a specific volume by ID:
+
+```bash
+remote snapshot create -v vol-12345678 -n my-snapshot
 ```
 
 List snapshots:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "remotepy"
-version = "1.5.0"
+version = "1.6.0"
 description = "CLI tool for managing AWS EC2 instances, ECS services, and related resources"
 authors = [{name = "Matthew Upson", email = "matt@mattupson.com"}]
 license = {text = "MIT License"}

remote/snapshot.py

@@ -1,5 +1,6 @@
 import typer
 
+from remote.exceptions import AWSServiceError
 from remote.instance_resolver import resolve_instance_or_exit
 from remote.utils import (
     confirm_action,
@@ -8,8 +9,10 @@
     get_ec2_client,
     get_status_style,
     get_volume_ids,
+    get_volumes_for_instance,
     handle_aws_errors,
     handle_cli_errors,
+    print_error,
     print_success,
     print_warning,
     styled_column,
@@ -19,10 +22,50 @@
 app = typer.Typer()
 
 
+def _get_device_name(volume: dict, instance_id: str) -> str:
+    """Extract the device name for a volume's attachment to a specific instance.
+
+    Args:
+        volume: Volume dictionary from describe_volumes
+        instance_id: The instance ID to match against
+
+    Returns:
+        The device name (e.g., "/dev/sdf") or empty string if not found
+    """
+    for attachment in volume.get("Attachments", []):
+        if attachment.get("InstanceId") == instance_id:
+            return attachment.get("Device", "")
+    return ""
+
+
+def _make_snapshot_name(base_name: str, device: str) -> str:
+    """Create a snapshot name with a device suffix.
+
+    Strips the /dev/ prefix and replaces slashes with dashes.
+
+    Args:
+        base_name: The base snapshot name
+        device: The device path (e.g., "/dev/sdf")
+
+    Returns:
+        Name with device suffix (e.g., "my-snapshot-sdf")
+    """
+    suffix = device.replace("/dev/", "").replace("/", "-")
+    return f"{base_name}-{suffix}" if suffix else base_name
+
+
 @app.command()
 @handle_cli_errors
 def create(
-    volume_id: str = typer.Option(..., "--volume-id", "-v", help="Volume ID (required)"),
+    instance_name: str | None = typer.Argument(
+        None, help="Instance name (auto-detects attached volumes)"
+    ),
+    volume_id: str | None = typer.Option(
+        None, "--volume-id", "-v", help="Volume ID to snapshot (alternative to instance name)"
+    ),
+    device: str | None = typer.Option(
+        None, "--device", "-D", help="Device filter when using instance name (e.g., /dev/sdf)"
+    ),
     name: str = typer.Option(..., "--name", "-n", help="Snapshot name (required)"),
     description: str = typer.Option("", "--description", "-d", help="Description"),
     yes: bool = typer.Option(
@@ -33,36 +76,123 @@ def create(
     ),
 ) -> None:
     """
-    Create an EBS snapshot from a volume.
+    Create EBS snapshot(s) from a volume or instance.
+
+    Provide an instance name to auto-detect and snapshot all attached volumes,
+    or use --volume-id for a single specific volume.
 
     Prompts for confirmation before creating.
 
     Examples:
+        remote snapshot create my-instance -n my-snapshot
+        remote snapshot create my-instance --device /dev/sdf -n my-snapshot
         remote snapshot create -v vol-123456 -n my-snapshot
         remote snapshot create -v vol-123456 -n backup -d "Daily backup"
         remote snapshot create -v vol-123456 -n backup --yes  # Skip confirmation
     """
-    validate_volume_id(volume_id)
-
-    # Confirm snapshot creation
-    if not yes:
-        if not confirm_action("create", "snapshot", name, details=f"from volume {volume_id}"):
-            print_warning("Snapshot creation cancelled")
-            return
-
-    with handle_aws_errors("EC2", "create_snapshot"):
-        snapshot = get_ec2_client().create_snapshot(
-            VolumeId=volume_id,
-            Description=description,
-            TagSpecifications=[
-                {
-                    "ResourceType": "snapshot",
-                    "Tags": [{"Key": "Name", "Value": name}],
-                }
-            ],
-        )
-        validate_aws_response_structure(snapshot, ["SnapshotId"], "create_snapshot")
-    print_success(f"Snapshot {snapshot['SnapshotId']} created")
+    # Validate option combinations
+    if device and not instance_name:
+        print_error("Error: --device can only be used with an instance name")
+        raise typer.Exit(1)
+    if volume_id and instance_name:
+        print_error("Error: --volume-id and instance name are mutually exclusive")
+        raise typer.Exit(1)
+    if not volume_id and not instance_name:
+        print_error("Error: Either an instance name or --volume-id is required")
+        raise typer.Exit(1)
+
+    if volume_id:
+        # Single-volume path
+        validate_volume_id(volume_id)
+
+        if not yes:
+            if not confirm_action("create", "snapshot", name, details=f"from volume {volume_id}"):
+                print_warning("Snapshot creation cancelled")
+                return
+
+        with handle_aws_errors("EC2", "create_snapshot"):
+            snapshot = get_ec2_client().create_snapshot(
+                VolumeId=volume_id,
+                Description=description,
+                TagSpecifications=[
+                    {
+                        "ResourceType": "snapshot",
+                        "Tags": [{"Key": "Name", "Value": name}],
+                    }
+                ],
+            )
+            validate_aws_response_structure(snapshot, ["SnapshotId"], "create_snapshot")
+        print_success(f"Snapshot {snapshot['SnapshotId']} created")
+    else:
+        # Instance-based path: auto-detect volumes
+        resolved_name, instance_id = resolve_instance_or_exit(instance_name)
+        volumes = get_volumes_for_instance(instance_id)
+
+        if not volumes:
+            print_error(f"Error: No volumes attached to instance {resolved_name}")
+            raise typer.Exit(1)
+
+        # Filter by device if specified
+        if device:
+            volumes = [v for v in volumes if _get_device_name(v, instance_id) == device]
+            if not volumes:
+                print_error(
+                    f"Error: No volume with device {device} attached to instance {resolved_name}"
+                )
+                raise typer.Exit(1)
+
+        # Build list of (volume_id, device, snapshot_name) tuples
+        snapshot_targets = []
+        for vol in volumes:
+            vid = vol["VolumeId"]
+            dev = _get_device_name(vol, instance_id)
+            if len(volumes) == 1:
+                snap_name = name
+            else:
+                snap_name = _make_snapshot_name(name, dev) if dev else f"{name}-{vid}"
+            snapshot_targets.append((vid, dev, snap_name))
+
+        # Confirm
+        if not yes:
+            details_lines = ", ".join(
+                f"{vid} ({dev})" if dev else vid for vid, dev, _ in snapshot_targets
+            )
+            if not confirm_action(
+                "create",
+                "snapshot(s) for instance",
+                resolved_name,
+                details=f"from volumes: {details_lines}",
+            ):
+                print_warning("Snapshot creation cancelled")
+                return
+
+        # Create snapshots, tracking successes and failures
+        created = []
+        failed = []
+        for vid, dev, snap_name in snapshot_targets:
+            dev_info = f" ({dev})" if dev else ""
+            try:
+                with handle_aws_errors("EC2", "create_snapshot"):
+                    snapshot = get_ec2_client().create_snapshot(
+                        VolumeId=vid,
+                        Description=description,
+                        TagSpecifications=[
+                            {
+                                "ResourceType": "snapshot",
+                                "Tags": [{"Key": "Name", "Value": snap_name}],
+                            }
+                        ],
+                    )
+                    validate_aws_response_structure(snapshot, ["SnapshotId"], "create_snapshot")
+                print_success(f"Snapshot {snapshot['SnapshotId']} created from {vid}{dev_info}")
+                created.append(vid)
+            except AWSServiceError as e:
+                print_error(f"Failed to create snapshot from {vid}{dev_info}: {e}")
+                failed.append(vid)
+
+        if failed:
+            print_warning(f"{len(created)} snapshot(s) created, {len(failed)} failed")
+            raise typer.Exit(1)
 
 
 @app.command("ls")

remote/utils.py

@@ -963,36 +963,44 @@ def get_instance_type(instance_id: str) -> str:
         raise
 
 
-def get_volume_ids(instance_id: str) -> list[str]:
-    """Returns a list of volume ids attached to the instance.
+def get_volumes_for_instance(instance_id: str) -> list[dict[str, Any]]:
+    """Returns volumes attached to the instance with full metadata.
+
+    Returns the full volume dictionaries including attachment info
+    (device names, state, etc.).
 
     Args:
         instance_id: The instance ID to get volumes for
 
     Returns:
-        List of volume IDs attached to the instance
+        List of volume dictionaries from describe_volumes
 
     Raises:
         AWSServiceError: If AWS API call fails
     """
-    # Validate input
     instance_id = validate_instance_id(instance_id)
 
     with handle_aws_errors("EC2", "describe_volumes"):
         response = get_ec2_client().describe_volumes(
             Filters=[{"Name": "attachment.instance-id", "Values": [instance_id]}]
         )
-
-        # Validate response structure
         validate_aws_response_structure(response, ["Volumes"], "describe_volumes")
+        return list(response["Volumes"])
+
 
-        # Safely extract volume IDs
-        volume_ids = []
-        for volume in response["Volumes"]:
-            if "VolumeId" in volume:
-                volume_ids.append(volume["VolumeId"])
+def get_volume_ids(instance_id: str) -> list[str]:
+    """Returns a list of volume ids attached to the instance.
 
-        return volume_ids
+    Args:
+        instance_id: The instance ID to get volumes for
+
+    Returns:
+        List of volume IDs attached to the instance
+
+    Raises:
+        AWSServiceError: If AWS API call fails
+    """
+    return [v["VolumeId"] for v in get_volumes_for_instance(instance_id) if "VolumeId" in v]
 
 
 def get_volume_name(volume_id: str) -> str: