1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
|
"""PodmanResource manager subclassed for Network resources.
Classes and methods for manipulating network resources via Podman API service.
Example:
with PodmanClient(base_url="unix:///run/user/1000/podman/podman.sock") as client:
for net in client.networks.list():
print(net.id, "\n")
"""
import ipaddress
import logging
from contextlib import suppress
from typing import Any, Optional, Literal
from podman.api import http_utils, prepare_filters
from podman.domain.manager import Manager
from podman.domain.networks import Network
from podman.errors import APIError
logger = logging.getLogger("podman.networks")
class NetworksManager(Manager):
"""Specialized Manager for Network resources."""
@property
def resource(self):
"""Type[Network]: prepare_model() will create Network classes."""
return Network
def create(self, name: str, **kwargs) -> Network:
"""Create a Network resource.
Args:
name: Name of network to be created
Keyword Args:
attachable (bool): Ignored, always False.
check_duplicate (bool): Ignored, always False.
dns_enabled (bool): When True, do not provision DNS for this network.
driver (str): Which network driver to use when creating network.
enable_ipv6 (bool): Enable IPv6 on the network.
ingress (bool): Ignored, always False.
internal (bool): Restrict external access to the network.
ipam (IPAMConfig): Optional custom IP scheme for the network.
labels (dict[str, str]): Map of labels to set on the network.
options (dict[str, Any]): Driver options.
scope (str): Ignored, always "local".
Raises:
APIError: when Podman service reports an error
"""
data = {
"name": name,
"driver": kwargs.get("driver"),
"dns_enabled": kwargs.get("dns_enabled"),
"subnets": kwargs.get("subnets"),
"ipv6_enabled": kwargs.get("enable_ipv6"),
"internal": kwargs.get("internal"),
"labels": kwargs.get("labels"),
"options": kwargs.get("options"),
}
with suppress(KeyError):
self._prepare_ipam(data, kwargs["ipam"])
response = self.client.post(
"/networks/create",
data=http_utils.prepare_body(data),
headers={"Content-Type": "application/json"},
)
response.raise_for_status()
return self.prepare_model(attrs=response.json())
def _prepare_ipam(self, data: dict[str, Any], ipam: dict[str, Any]):
if "Driver" in ipam:
data["ipam_options"] = {"driver": ipam["Driver"]}
if "Config" not in ipam:
return
data["subnets"] = []
for cfg in ipam["Config"]:
subnet = {
"gateway": cfg.get("Gateway"),
"subnet": cfg.get("Subnet"),
}
with suppress(KeyError):
net = ipaddress.ip_network(cfg["IPRange"])
subnet["lease_range"] = {
"start_ip": str(net[1]),
"end_ip": str(net[-2]),
}
data["subnets"].append(subnet)
def exists(self, key: str) -> bool:
response = self.client.get(f"/networks/{key}/exists")
return response.ok
def get(self, key: str) -> Network:
"""Return information for the network_id.
Args:
key: Network name or id.
Raises:
NotFound: when Network does not exist
APIError: when error returned by service
"""
response = self.client.get(f"/networks/{key}")
response.raise_for_status()
return self.prepare_model(attrs=response.json())
def list(self, **kwargs) -> list[Network]:
"""Report on networks.
Keyword Args:
names (list[str]): List of names to filter by.
ids (list[str]): List of identifiers to filter by.
filters (Mapping[str,str]): Criteria for listing networks. Available filters:
- driver="bridge": Matches a network's driver. Only "bridge" is supported.
- label=(Union[str, list[str]]): format either "key", "key=value"
or a list of such.
- type=(str): Filters networks by type, legal values are:
- "custom"
- "builtin"
- plugin=(list[str]]): Matches CNI plugins included in a network, legal
values are (Podman only):
- bridge
- portmap
- firewall
- tuning
- dnsname
- macvlan
greedy (bool): Fetch more details for each network individually.
You might want this to get the containers attached to them. Ignored.
Raises:
APIError: when error returned by service
"""
filters = kwargs.get("filters", {})
filters["name"] = kwargs.get("names")
filters["id"] = kwargs.get("ids")
filters = prepare_filters(filters)
params = {"filters": filters}
response = self.client.get("/networks/json", params=params)
response.raise_for_status()
return [self.prepare_model(i) for i in response.json()]
def prune(
self, filters: Optional[dict[str, Any]] = None
) -> dict[Literal["NetworksDeleted", "SpaceReclaimed"], Any]:
"""Delete unused Networks.
SpaceReclaimed always reported as 0
Args:
filters: Criteria for selecting volumes to delete. Ignored.
Raises:
APIError: when service reports error
"""
params = {"filters": prepare_filters(filters)}
response = self.client.post("/networks/prune", params=params)
response.raise_for_status()
deleted: list[str] = []
for item in response.json():
if item["Error"] is not None:
raise APIError(
item["Error"],
response=response,
explanation=f"""Failed to prune network '{item["Name"]}'""",
)
deleted.append(item["Name"])
return {"NetworksDeleted": deleted, "SpaceReclaimed": 0}
def remove(self, name: [Network, str], force: Optional[bool] = None) -> None:
"""Remove Network resource.
Args:
name: Identifier of Network to delete.
force: Remove network and any associated containers
Raises:
APIError: when Podman service reports an error
"""
if isinstance(name, Network):
name = name.name
response = self.client.delete(f"/networks/{name}", params={"force": force})
response.raise_for_status()
|
