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
|
"""PodmanResource manager subclassed for Containers."""
import logging
import urllib
from typing import Any, Union
from collections.abc import Mapping
from podman import api
from podman.domain.containers import Container
from podman.domain.containers_create import CreateMixin
from podman.domain.containers_run import RunMixin
from podman.domain.manager import Manager
from podman.errors import APIError
logger = logging.getLogger("podman.containers")
class ContainersManager(RunMixin, CreateMixin, Manager):
"""Specialized Manager for Container resources."""
@property
def resource(self):
"""Type[Container]: prepare_model() will create Container classes."""
return Container
def exists(self, key: str) -> bool:
response = self.client.get(f"/containers/{key}/exists")
return response.ok
def get(self, key: str) -> Container:
"""Get container by name or id.
Args:
key: Container name or id.
Returns:
A `Container` object corresponding to `key`.
Raises:
NotFound: when Container does not exist
APIError: when an error return by service
"""
container_id = urllib.parse.quote_plus(key)
response = self.client.get(f"/containers/{container_id}/json")
response.raise_for_status()
return self.prepare_model(attrs=response.json())
def list(self, **kwargs) -> list[Container]:
"""Report on containers.
Keyword Args:
all: If False, only show running containers. Default: False.
since: Show containers created after container name or id given.
before: Show containers created before container name or id given.
limit: Show last N created containers.
filters: Filter container reported.
Available filters:
- exited (int): Only containers with specified exit code
- status (str): One of restarting, running, paused, exited
- label (Union[str, list[str]]): Format either "key", "key=value" or a list of such.
- id (str): The id of the container.
- name (str): The name of the container.
- ancestor (str): Filter by container ancestor. Format of
<image-name>[:tag], <image-id>, or <image@digest>.
- before (str): Only containers created before a particular container.
Give the container name or id.
- since (str): Only containers created after a particular container.
Give container name or id.
sparse: Ignored
ignore_removed: If True, ignore failures due to missing containers.
Raises:
APIError: when service returns an error
"""
params = {
"all": kwargs.get("all"),
"filters": kwargs.get("filters", {}),
"limit": kwargs.get("limit"),
}
if "before" in kwargs:
params["filters"]["before"] = kwargs.get("before")
if "since" in kwargs:
params["filters"]["since"] = kwargs.get("since")
# filters formatted last because some kwargs may need to be mapped into filters
params["filters"] = api.prepare_filters(params["filters"])
response = self.client.get("/containers/json", params=params)
response.raise_for_status()
return [self.prepare_model(attrs=i) for i in response.json()]
def prune(self, filters: Mapping[str, str] = None) -> dict[str, Any]:
"""Delete stopped containers.
Args:
filters: Criteria for determining containers to remove. Available keys are:
- until (str): Delete containers before this time
- label (list[str]): Labels associated with containers
Returns:
Keys:
- ContainersDeleted (list[str]): Identifiers of deleted containers.
- SpaceReclaimed (int): Amount of disk space reclaimed in bytes.
Raises:
APIError: when service reports an error
"""
params = {"filters": api.prepare_filters(filters)}
response = self.client.post("/containers/prune", params=params)
response.raise_for_status()
results = {"ContainersDeleted": [], "SpaceReclaimed": 0}
for entry in response.json():
if entry.get("Err") is not None:
raise APIError(
entry["Err"],
response=response,
explanation=f"""Failed to prune container '{entry["Id"]}'""",
)
results["ContainersDeleted"].append(entry["Id"])
results["SpaceReclaimed"] += entry["Size"]
return results
def remove(self, container_id: Union[Container, str], **kwargs):
"""Delete container.
Podman only
Args:
container_id: identifier of Container to delete.
Keyword Args:
v (bool): Delete associated volumes as well.
link (bool): Ignored.
force (bool): Kill a running container before deleting.
"""
if isinstance(container_id, Container):
container_id = container_id.id
# v is used for the compat endpoint while volumes is used for the libpod endpoint
params = {"v": kwargs.get("v"), "force": kwargs.get("force"), "volumes": kwargs.get("v")}
response = self.client.delete(f"/containers/{container_id}", params=params)
response.raise_for_status()
|
