feat(purge): Add ability to purge single service

devservices/commands/purge.py

@@ -6,9 +6,15 @@
 from argparse import ArgumentParser
 from argparse import Namespace
 
+from devservices.configs.service_config import load_service_config_from_file
+from devservices.constants import DEPENDENCY_CONFIG_VERSION
 from devservices.constants import DEVSERVICES_CACHE_DIR
+from devservices.constants import DEVSERVICES_DEPENDENCIES_CACHE_DIR
 from devservices.constants import DEVSERVICES_ORCHESTRATOR_LABEL
 from devservices.constants import DOCKER_NETWORK_NAME
+from devservices.exceptions import ConfigNotFoundError
+from devservices.exceptions import ConfigParseError
+from devservices.exceptions import ConfigValidationError
 from devservices.exceptions import DockerDaemonNotRunningError
 from devservices.exceptions import DockerError
 from devservices.utils.console import Console
@@ -19,17 +25,127 @@
 from devservices.utils.docker import remove_docker_resources
 from devservices.utils.docker import stop_containers
 from devservices.utils.state import State
+from devservices.utils.state import StateTables
 
 
 def add_parser(subparsers: _SubParsersAction[ArgumentParser]) -> None:
     parser = subparsers.add_parser("purge", help="Purge the local devservices cache")
+    parser.add_argument(
+        "service_name",
+        nargs="?",
+        help="Service name to purge (optional, purges all if not specified)",
+        default=None,
+    )
     parser.set_defaults(func=purge)
 
 
-def purge(_args: Namespace) -> None:
+def _get_service_cache_paths(service_name: str) -> list[str]:
+    """Find cache directory paths for a given service name."""
+
+    cache_paths: list[str] = []
+    dependencies_cache_dir = os.path.join(
+        DEVSERVICES_DEPENDENCIES_CACHE_DIR, DEPENDENCY_CONFIG_VERSION
+    )
+
+    if not os.path.exists(dependencies_cache_dir):
+        return cache_paths
+
+    for repo_name in os.listdir(dependencies_cache_dir):
+        repo_path = os.path.join(dependencies_cache_dir, repo_name)
+        if not os.path.isdir(repo_path):
+            continue
+
+        try:
+            service_config = load_service_config_from_file(repo_path)
+            if service_config.service_name == service_name:
+                cache_paths.append(repo_path)
+        except (ConfigNotFoundError, ConfigParseError, ConfigValidationError):
+            # Skip invalid configs
+            continue
+
+    return cache_paths
+
+
+def purge(args: Namespace) -> None:
     """Purge the local devservices state and cache and remove all devservices containers and volumes."""
     console = Console()
+    service_name = getattr(args, "service_name", None)
+
+    if service_name:
+        _purge_service(service_name, console)
+    else:
+        _purge_all(console)
+
+
+def _purge_service(service_name: str, console: Console) -> None:
+    """Purge a specific service."""
+    state = State()
+
+    state.remove_service_entry(service_name, StateTables.STARTED_SERVICES)
+    state.remove_service_entry(service_name, StateTables.STARTING_SERVICES)
+    state.remove_service_entry(service_name, StateTables.SERVICE_RUNTIME)
+
+    try:
+        service_containers = get_matching_containers(
+            [
+                DEVSERVICES_ORCHESTRATOR_LABEL,
+                f"com.docker.compose.service={service_name}",
+            ]
+        )
+    except DockerDaemonNotRunningError as e:
+        console.warning(str(e))
+        service_containers = []
+    except DockerError as de:
+        console.failure(f"Failed to get containers for {service_name}: {de.stderr}")
+        exit(1)
+
+    if len(service_containers) == 0:
+        console.warning(f"No containers found for {service_name}")
+    else:
+        try:
+            service_volumes = get_volumes_for_containers(service_containers)
+        except DockerError as e:
+            console.failure(f"Failed to get volumes for {service_name}: {e.stderr}")
+            exit(1)
+
+        with Status(
+            lambda: console.warning(f"Stopping {service_name} containers"),
+            lambda: console.success(f"{service_name} containers have been stopped"),
+        ):
+            try:
+                stop_containers(service_containers, should_remove=True)
+            except DockerError as e:
+                console.failure(f"Failed to stop {service_name} containers: {e.stderr}")
+                exit(1)
+
+        console.warning(f"Removing {service_name} docker volumes")
+        if len(service_volumes) == 0:
+            console.success(f"No volumes found for {service_name}")
+        else:
+            try:
+                remove_docker_resources("volume", list(service_volumes))
+                console.success(f"{service_name} volumes removed")
+            except DockerError as e:
+                console.failure(f"Failed to remove {service_name} volumes: {e.stderr}")
+
+    cache_paths = _get_service_cache_paths(service_name)
+    if cache_paths:
+        console.warning(f"Removing cache for {service_name}")
+        for cache_path in cache_paths:
+            try:
+                shutil.rmtree(cache_path)
+            except PermissionError as e:
+                console.failure(f"Failed to remove cache at {cache_path}: {e}")
+                exit(1)
+        console.success(f"Cache for {service_name} has been removed")
+    else:
+        console.success(f"No cache found for {service_name}")
+
+    console.success(f"{service_name} has been purged")
+
 
+def _purge_all(console: Console) -> None:
+    """Purge all devservices state, cache, containers, volumes, and networks."""
     if os.path.exists(DEVSERVICES_CACHE_DIR):
         try:
             shutil.rmtree(DEVSERVICES_CACHE_DIR)

tests/commands/test_purge.py

@@ -552,3 +552,101 @@ def test_purge_with_cache_and_state_and_containers_with_networks_and_volumes(
                 mock.call("network", ["abc", "def", "ghe"]),
             ]
         )
+
+
+@mock.patch("devservices.commands.purge.get_matching_containers")
+@mock.patch("devservices.commands.purge.get_volumes_for_containers")
+@mock.patch("devservices.commands.purge.stop_containers")
+@mock.patch("devservices.commands.purge.remove_docker_resources")
+@mock.patch("devservices.commands.purge._get_service_cache_paths")
+def test_purge_specific_service(
+    mock_get_service_cache_paths: mock.Mock,
+    mock_remove_docker_resources: mock.Mock,
+    mock_stop_containers: mock.Mock,
+    mock_get_volumes_for_containers: mock.Mock,
+    mock_get_matching_containers: mock.Mock,
+    tmp_path: Path,
+) -> None:
+    """Test that purging a specific service removes only that service's containers, volumes, and cache."""
+    mock_get_matching_containers.return_value = [
+        "kafka-container-1",
+        "kafka-container-2",
+    ]
+    mock_get_volumes_for_containers.return_value = ["kafka-volume-1", "kafka-volume-2"]
+    cache_path = tmp_path / "dependencies" / "v1" / "kafka-repo"
+    cache_path.mkdir(parents=True, exist_ok=True)
+    mock_get_service_cache_paths.return_value = [str(cache_path)]
+
+    with (
+        mock.patch("devservices.utils.state.STATE_DB_FILE", str(tmp_path / "state")),
+        mock.patch(
+            "devservices.utils.docker.check_docker_daemon_running", return_value=None
+        ),
+    ):
+        state = State()
+        state.update_service_entry("kafka", "default", StateTables.STARTED_SERVICES)
+        state.update_service_entry("redis", "default", StateTables.STARTED_SERVICES)
+
+        assert state.get_service_entries(StateTables.STARTED_SERVICES) == [
+            "kafka",
+            "redis",
+        ]
+        assert cache_path.exists()
+
+        purge(Namespace(service_name="kafka"))
+
+        # Only kafka should be removed from state
+        assert state.get_service_entries(StateTables.STARTED_SERVICES) == ["redis"]
+        # Cache path should be removed
+        assert not cache_path.exists()
+
+        # Should filter containers by service name
+        mock_get_matching_containers.assert_called_once_with(
+            [
+                DEVSERVICES_ORCHESTRATOR_LABEL,
+                "com.docker.compose.service=kafka",
+            ]
+        )
+        mock_get_volumes_for_containers.assert_called_once_with(
+            ["kafka-container-1", "kafka-container-2"]
+        )
+        mock_stop_containers.assert_called_once_with(
+            ["kafka-container-1", "kafka-container-2"], should_remove=True
+        )
+        mock_remove_docker_resources.assert_called_once_with(
+            "volume", ["kafka-volume-1", "kafka-volume-2"]
+        )
+        mock_get_service_cache_paths.assert_called_once_with("kafka")
+
+
+@mock.patch("devservices.commands.purge.get_matching_containers")
+@mock.patch("devservices.commands.purge._get_service_cache_paths")
+def test_purge_specific_service_no_containers(
+    mock_get_service_cache_paths: mock.Mock,
+    mock_get_matching_containers: mock.Mock,
+    capsys: pytest.CaptureFixture[str],
+    tmp_path: Path,
+) -> None:
+    """Test that purging a service with no containers or cache still removes it from state."""
+    mock_get_matching_containers.return_value = []
+    mock_get_service_cache_paths.return_value = []
+
+    with (
+        mock.patch("devservices.utils.state.STATE_DB_FILE", str(tmp_path / "state")),
+        mock.patch(
+            "devservices.utils.docker.check_docker_daemon_running", return_value=None
+        ),
+    ):
+        state = State()
+        state.update_service_entry("kafka", "default", StateTables.STARTED_SERVICES)
+
+        args = Namespace(service_name="kafka")
+        purge(args)
+
+        # Service should be removed from state even if no containers found
+        assert state.get_service_entries(StateTables.STARTED_SERVICES) == []
+
+        captured = capsys.readouterr()
+        assert "No containers found for kafka" in captured.out
+        assert "No cache found for kafka" in captured.out
+        assert "kafka has been purged" in captured.out