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
|
"""Model for Network resources.
Example:
with PodmanClient(base_url="unix:///run/user/1000/podman/podman.sock") as client:
net = client.networks.get("db_network")
print(net.name, "\n")
"""
import hashlib
import json
import logging
from contextlib import suppress
from typing import Optional, Union
from podman.domain.containers import Container
from podman.domain.containers_manager import ContainersManager
from podman.domain.manager import PodmanResource
logger = logging.getLogger("podman.networks")
class Network(PodmanResource):
"""Details and configuration for a networks managed by the Podman service.
Attributes:
attrs (dict[str, Any]): Attributes of Network reported from Podman service
"""
@property
def id(self): # pylint: disable=invalid-name
"""str: Returns the identifier of the network."""
with suppress(KeyError):
return self.attrs["Id"]
with suppress(KeyError):
sha256 = hashlib.sha256(self.attrs["name"].encode("ascii"))
return sha256.hexdigest()
return None
@property
def containers(self):
"""list[Container]: Returns list of Containers connected to network."""
with suppress(KeyError):
container_manager = ContainersManager(client=self.client)
return [container_manager.get(ident) for ident in self.attrs["Containers"].keys()]
return []
@property
def name(self):
"""str: Returns the name of the network."""
if "Name" in self.attrs:
return self.attrs["Name"]
if "name" in self.attrs:
return self.attrs["name"]
raise KeyError("Neither 'name' or 'Name' attribute found.")
def reload(self):
"""Refresh this object's data from the service."""
latest = self.manager.get(self.name)
self.attrs = latest.attrs
def connect(self, container: Union[str, Container], *_, **kwargs) -> None:
"""Connect given container to this network.
Args:
container: To add to this Network
Keyword Args:
aliases (list[str]): Aliases to add for this endpoint
driver_opt (dict[str, Any]): Options to provide to network driver
ipv4_address (str): IPv4 address for given Container on this network
ipv6_address (str): IPv6 address for given Container on this network
link_local_ips (list[str]): list of link-local addresses
links (list[Union[str, Containers]]): Ignored
Raises:
APIError: when Podman service reports an error
"""
if isinstance(container, Container):
container = container.id
# TODO Talk with baude on which IPAddress field is needed...
ipam = {
"IPv4Address": kwargs.get('ipv4_address'),
"IPv6Address": kwargs.get('ipv6_address'),
"Links": kwargs.get("link_local_ips"),
}
ipam = {k: v for (k, v) in ipam.items() if not (v is None or len(v) == 0)}
endpoint_config = {
"Aliases": kwargs.get("aliases"),
"DriverOpts": kwargs.get("driver_opt"),
"IPAddress": kwargs.get("ipv4_address", kwargs.get("ipv6_address")),
"IPAMConfig": ipam,
"Links": kwargs.get("link_local_ips"),
"NetworkID": self.id,
}
endpoint_config = {
k: v for (k, v) in endpoint_config.items() if not (v is None or len(v) == 0)
}
data = {"Container": container, "EndpointConfig": endpoint_config}
data = {k: v for (k, v) in data.items() if not (v is None or len(v) == 0)}
response = self.client.post(
f"/networks/{self.name}/connect",
data=json.dumps(data),
headers={"Content-type": "application/json"},
**kwargs,
)
response.raise_for_status()
def disconnect(self, container: Union[str, Container], **kwargs) -> None:
"""Disconnect given container from this network.
Args:
container: To remove from this Network
Keyword Args:
force (bool): Force operation
Raises:
APIError: when Podman service reports an error
"""
if isinstance(container, Container):
container = container.id
data = {"Container": container, "Force": kwargs.get("force")}
response = self.client.post(f"/networks/{self.name}/disconnect", data=json.dumps(data))
response.raise_for_status()
def remove(self, force: Optional[bool] = None, **kwargs) -> None:
"""Remove this network.
Args:
force: Remove network and any associated containers
Raises:
APIError: when Podman service reports an error
"""
self.manager.remove(self.name, force=force, **kwargs)
|
