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
|
"""Model and Manager for Volume resources."""
import logging
from typing import Any, Literal, Optional, Union
import requests
from podman import api
from podman.domain.manager import Manager, PodmanResource
from podman.errors import APIError
logger = logging.getLogger("podman.volumes")
class Volume(PodmanResource):
"""Details and configuration for an image managed by the Podman service."""
@property
def id(self):
return self.name
@property
def name(self):
"""str: Returns the name of the volume."""
return self.attrs.get("Name")
def remove(self, force: Optional[bool] = None) -> None:
"""Delete this volume.
Args:
force: When true, force deletion of in-use volume
Raises:
APIError: when service reports an error
"""
self.manager.remove(self.name, force=force)
class VolumesManager(Manager):
"""Specialized Manager for Volume resources."""
@property
def resource(self):
"""Type[Volume]: prepare_model() will create Volume classes."""
return Volume
def create(self, name: Optional[str] = None, **kwargs) -> Volume:
"""Create a Volume.
Args:
name: Name given to new volume
Keyword Args:
driver (str): Volume driver to use
driver_opts (dict[str, str]): Options to use with driver
labels (dict[str, str]): Labels to apply to volume
Raises:
APIError: when service reports error
"""
data = {
"Driver": kwargs.get("driver"),
"Labels": kwargs.get("labels"),
"Name": name,
"Options": kwargs.get("driver_opts"),
}
response = self.client.post(
"/volumes/create",
data=api.prepare_body(data),
headers={"Content-Type": "application/json"},
)
response.raise_for_status()
return self.prepare_model(attrs=response.json())
def exists(self, key: str) -> bool:
response = self.client.get(f"/volumes/{key}/exists")
return response.ok
# pylint is flagging 'volume_id' here vs. 'key' parameter in super.get()
def get(self, volume_id: str) -> Volume: # pylint: disable=arguments-differ,arguments-renamed
"""Returns and volume by name or id.
Args:
volume_id: Volume id or name for which to search
Raises:
NotFound: when volume could not be found
APIError: when service reports an error
"""
response = self.client.get(f"/volumes/{volume_id}/json")
response.raise_for_status()
return self.prepare_model(attrs=response.json())
def list(self, *_, **kwargs) -> list[Volume]:
"""Report on volumes.
Keyword Args:
filters (dict[str, str]): criteria to filter Volume list
- driver (str): filter volumes by their driver
- label (dict[str, str]): filter by label and/or value
- name (str): filter by volume's name
"""
filters = api.prepare_filters(kwargs.get("filters"))
response = self.client.get("/volumes/json", params={"filters": filters})
if response.status_code == requests.codes.not_found:
return []
response.raise_for_status()
return [self.prepare_model(i) for i in response.json()]
def prune(
self,
filters: Optional[dict[str, str]] = None, # pylint: disable=unused-argument
) -> dict[Literal["VolumesDeleted", "SpaceReclaimed"], Any]:
"""Delete unused volumes.
Args:
filters: Criteria for selecting volumes to delete. Ignored.
Raises:
APIError: when service reports error
"""
response = self.client.post("/volumes/prune")
data = response.json()
response.raise_for_status()
volumes: list[str] = []
space_reclaimed = 0
for item in data:
if "Err" in item:
raise APIError(
item["Err"],
response=response,
explanation=f"""Failed to prune volume '{item.get("Id")}'""",
)
volumes.append(item.get("Id"))
space_reclaimed += item["Size"]
return {"VolumesDeleted": volumes, "SpaceReclaimed": space_reclaimed}
def remove(self, name: Union[Volume, str], force: Optional[bool] = None) -> None:
"""Delete a volume.
Podman only.
Args:
name: Identifier for Volume to be deleted.
force: When true, force deletion of in-use volume
Raises:
APIError: when service reports an error
"""
if isinstance(name, Volume):
name = name.name
response = self.client.delete(f"/volumes/{name}", params={"force": force})
response.raise_for_status()
|
