Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGES/691.feature.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Introduce support in the library for the `networks/prune` API endpoint.
26 changes: 26 additions & 0 deletions aiodocker/networks.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,32 @@ async def get(self, net_specs: str) -> DockerNetwork:
data = await self.docker._query_json(f"networks/{net_specs}", method="GET")
return DockerNetwork(self.docker, data["Id"])

async def prune(
self,
*,
filters: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
"""
Delete unused networks

Args:
filters: Filter expressions to limit which networks are pruned.
Available filters:
- until: Only remove networks created before given timestamp
- label: Only remove networks with (or without, if label!=<key> is used) the specified labels

Returns:
Dictionary containing information about deleted networks
"""
params = {}
if filters is not None:
params["filters"] = clean_filters(filters)

response = await self.docker._query_json(
"networks/prune", "POST", params=params
)
return response


class DockerNetwork:
def __init__(self, docker, id_):
Expand Down
47 changes: 47 additions & 0 deletions tests/test_networks.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
from __future__ import annotations

from contextlib import suppress

import pytest

from aiodocker.docker import Docker
Expand Down Expand Up @@ -56,3 +58,48 @@ async def test_network_delete_error(docker: Docker) -> None:
assert await network.delete() is True
with pytest.raises(DockerError):
await network.delete()


@pytest.mark.asyncio
async def test_prune_networks(docker: Docker, random_name: str) -> None:
"""Test that prune with filters removes only the unused network that matches the filter."""
# Create two networks
network_without_label: DockerNetwork = await docker.networks.create({
"Name": f"{random_name}_no_label"
})
network_with_label: DockerNetwork | None = None
try:
network_name = f"{random_name}_label"
network_with_label = await docker.networks.create({
"Name": network_name,
"Labels": {"test": ""},
})

# Prune unused networks with label "test"
result = await docker.networks.prune(filters={"label": "test"})

# Verify the response structure
assert isinstance(result, dict)
assert "NetworksDeleted" in result
assert result["NetworksDeleted"] == [network_name]

# Test that the network without the label still exists
assert await network_without_label.show()

finally:
with suppress(DockerError):
await network_without_label.delete()
if network_with_label:
with suppress(DockerError):
await network_with_label.delete()


@pytest.mark.asyncio
async def test_prune_networks_nothing_to_remove(docker: Docker) -> None:
"""Test a network prune with nothing to remove."""
result = await docker.networks.prune()

# Verify the response structure
assert isinstance(result, dict)
assert "NetworksDeleted" in result
assert result["NetworksDeleted"] is None