feat: add support for containers/prune API

CHANGES/1002.feature.md

@@ -0,0 +1 @@
+Introduce support in the library for the `containers/prune` API endpoint.

aiodocker/containers.py

@@ -30,7 +30,7 @@
 from .multiplexed import multiplexed_result_list, multiplexed_result_stream
 from .stream import Stream
 from .types import SENTINEL, JSONObject, MutableJSONObject, PortInfo, Sentinel
-from .utils import identical, parse_result
+from .utils import clean_filters, identical, parse_result
 
 
 if TYPE_CHECKING:
@@ -136,6 +136,32 @@ def exec(self, exec_id: str) -> Exec:
         """Return Exec instance for already created exec object."""
         return Exec(self.docker, exec_id)
 
+    async def prune(
+        self,
+        *,
+        filters: Optional[Mapping[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Delete stopped containers
+
+        Args:
+            filters: Filter expressions to limit which containers are pruned.
+                    Available filters:
+                    - until: Only remove containers created before given timestamp
+                    - label: Only remove containers with (or without, if label!=<key> is used) the specified labels
+
+        Returns:
+            Dictionary containing information about deleted containers and space reclaimed
+        """
+        params = {}
+        if filters is not None:
+            params["filters"] = clean_filters(filters)
+
+        response = await self.docker._query_json(
+            "containers/prune", "POST", params=params
+        )
+        return response
+
 
 class DockerContainer:
     _container: Dict[str, Any]

tests/test_containers.py

@@ -1,6 +1,7 @@
 import asyncio
 import secrets
 import sys
+from contextlib import suppress
 
 import pytest
 
@@ -451,3 +452,53 @@ async def test_delete_running_container_with_force(
         # If container was already deleted, that's fine
         if e.status != 404:
             raise
+
+
+@pytest.mark.asyncio
+async def test_prune_containers(
+    docker: Docker, random_name: str, image_name: str
+) -> None:
+    """Test that prune with filters removes only the stopped container that matches the filter."""
+    # Create two stopped containers
+    container_without_label: DockerContainer = await docker.containers.create(
+        {"Image": image_name}, name=f"{random_name}_no_label"
+    )
+    container_with_label: DockerContainer | None = None
+    try:
+        container_with_label = await docker.containers.create(
+            {"Image": image_name, "Labels": {"test": ""}},
+            name=f"{random_name}_with_label",
+        )
+
+        # Prune stopped containers with label "test"
+        result = await docker.containers.prune(filters={"label": "test"})
+
+        # Verify the response structure
+        assert isinstance(result, dict)
+        assert "ContainersDeleted" in result
+        assert "SpaceReclaimed" in result
+        assert result["ContainersDeleted"] == [container_with_label.id]
+        assert isinstance(result["SpaceReclaimed"], int)
+
+        # Test that the container without the label still exists
+        assert await container_without_label.show()
+
+    finally:
+        with suppress(DockerError):
+            await container_without_label.delete()
+        if container_with_label:
+            with suppress(DockerError):
+                await container_with_label.delete()
+
+
+@pytest.mark.asyncio
+async def test_prune_containers_nothing_to_remove(docker: Docker) -> None:
+    """Test a container prune with nothing to remove."""
+    result = await docker.containers.prune()
+
+    # Verify the response structure
+    assert isinstance(result, dict)
+    assert "ContainersDeleted" in result
+    assert "SpaceReclaimed" in result
+    assert result["ContainersDeleted"] is None
+    assert isinstance(result["SpaceReclaimed"], int)