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
|
"""PodmanResource manager subclassed for Networks."""
import builtins
import json
import logging
from typing import Any, Optional, Union
from collections.abc import Iterator
from podman import api
from podman.domain.manager import Manager
from podman.domain.pods import Pod
from podman.errors import APIError
logger = logging.getLogger("podman.pods")
class PodsManager(Manager):
"""Specialized Manager for Pod resources."""
@property
def resource(self):
"""Type[Pod]: prepare_model() will create Pod classes."""
return Pod
def create(self, name: str, **kwargs) -> Pod:
"""Create a Pod.
Keyword Args:
See (API documentation)[
https://docs.podman.io/en/latest/_static/api.html#operation/CreatePod] for
complete list of keywords.
"""
data = {} if kwargs is None else kwargs.copy()
data["name"] = name
response = self.client.post("/pods/create", data=json.dumps(data))
response.raise_for_status()
body = response.json()
return self.get(body["Id"])
def exists(self, key: str) -> bool:
"""Returns True, when pod exists."""
response = self.client.get(f"/pods/{key}/exists")
return response.ok
# pylint is flagging 'pod_id' here vs. 'key' parameter in super.get()
def get(self, pod_id: str) -> Pod: # pylint: disable=arguments-differ,arguments-renamed
"""Return information for Pod by name or id.
Args:
pod_id: Pod name or id.
Raises:
NotFound: when network does not exist
APIError: when error returned by service
"""
response = self.client.get(f"/pods/{pod_id}/json")
response.raise_for_status()
return self.prepare_model(attrs=response.json())
def list(self, **kwargs) -> builtins.list[Pod]:
"""Report on pods.
Keyword Args:
filters (Mapping[str, str]): Criteria for listing pods. Available filters:
- ctr-ids (list[str]): list of container ids to filter by.
- ctr-names (list[str]): list of container names to filter by.
- ctr-number (list[int]): list pods with given number of containers.
- ctr-status (list[str]): list pods with containers in given state.
Legal values are: "created", "running", "paused", "stopped",
"exited", or "unknown"
- id (str) - List pod with this id.
- name (str) - List pod with this name.
- status (list[str]): List pods in given state. Legal values are:
"created", "running", "paused", "stopped", "exited", or "unknown"
- label (list[str]): List pods with given labels.
- network (list[str]): List pods associated with given Network Ids (not Names).
Raises:
APIError: when an error returned by service
"""
params = {"filters": api.prepare_filters(kwargs.get("filters"))}
response = self.client.get("/pods/json", params=params)
response.raise_for_status()
return [self.prepare_model(attrs=i) for i in response.json()]
def prune(self, filters: Optional[dict[str, str]] = None) -> dict[str, Any]:
"""Delete unused Pods.
Returns:
Dictionary Keys:
- PodsDeleted (list[str]): List of pod ids deleted.
- SpaceReclaimed (int): Always zero.
Raises:
APIError: when service reports error
"""
response = self.client.post("/pods/prune", params={"filters": api.prepare_filters(filters)})
response.raise_for_status()
deleted: builtins.list[str] = []
for item in response.json():
if item["Err"] is not None:
raise APIError(
item["Err"],
response=response,
explanation=f"""Failed to prune pod '{item["Id"]}'""",
)
deleted.append(item["Id"])
return {"PodsDeleted": deleted, "SpaceReclaimed": 0}
def remove(self, pod_id: Union[Pod, str], force: Optional[bool] = None) -> None:
"""Delete pod.
Args:
pod_id: Identifier of Pod to delete.
force: When True, stop and delete all containers in pod before deleting pod.
Raises:
NotFound: when pod not found
APIError: when service reports an error
Notes:
Podman only.
"""
if isinstance(pod_id, Pod):
pod_id = pod_id.id
response = self.client.delete(f"/pods/{pod_id}", params={"force": force})
response.raise_for_status()
def stats(
self, **kwargs
) -> Union[builtins.list[dict[str, Any]], Iterator[builtins.list[dict[str, Any]]]]:
"""Resource usage statistics for the containers in pods.
Keyword Args:
all (bool): Provide statistics for all running pods.
name (Union[str, list[str]]): Pods to include in report.
stream (bool): Stream statistics until cancelled. Default: False.
decode (bool): If True, response will be decoded into dict. Default: False.
Raises:
NotFound: when pod not found
APIError: when service reports an error
"""
if "all" in kwargs and "name" in kwargs:
raise ValueError("Keywords 'all' and 'name' are mutually exclusive.")
# Keeping the default for stream as False to not break existing users
# Should probably be changed in a newer major version to match behavior of container.stats
stream = kwargs.get("stream", False)
decode = kwargs.get("decode", False)
params = {
"all": kwargs.get("all"),
"namesOrIDs": kwargs.get("name"),
"stream": stream,
}
response = self.client.get("/pods/stats", params=params, stream=stream)
response.raise_for_status()
if stream:
return api.stream_helper(response, decode_to_json=decode)
return json.loads(response.content) if decode else response.content
|
