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
|
"""Model and Manager for Secrets resources."""
from contextlib import suppress
from typing import Any, Optional, Union
from collections.abc import Mapping
from podman.api import APIClient
from podman.domain.manager import Manager, PodmanResource
class Secret(PodmanResource):
"""Details and configuration for a secret registered with the Podman service."""
def __repr__(self):
return f"<{self.__class__.__name__}: {self.name}>"
@property
def id(self): # pylint: disable=invalid-name
return self.attrs.get("ID")
@property
def name(self):
"""str: name of the secret."""
with suppress(KeyError):
return self.attrs['Spec']['Name']
return ""
def remove(
self,
all: Optional[bool] = None, # pylint: disable=redefined-builtin
):
"""Delete secret.
Args:
all: When True, delete all secrets.
Raises:
NotFound: when Secret does not exist
APIError: when error returned by service
"""
self.manager.remove(self.id, all=all)
class SecretsManager(Manager):
"""Specialized Manager for Secret resources."""
@property
def resource(self):
"""Type[Secret]: prepare_model() will create Secret classes."""
return Secret
def __init__(self, client: APIClient):
"""Initialize SecretsManager object.
Args:
client: Connection to Podman service.
"""
super().__init__(client)
def exists(self, key: str) -> bool:
response = self.client.get(f"/secrets/{key}/json")
return response.ok
# pylint is flagging 'secret_id' here vs. 'key' parameter in super.get()
def get(self, secret_id: str) -> Secret: # pylint: disable=arguments-differ,arguments-renamed
"""Return information for Secret by name or id.
Args:
secret_id: Secret name or id.
Raises:
NotFound: when Secret does not exist
APIError: when error returned by service
"""
response = self.client.get(f"/secrets/{secret_id}/json")
response.raise_for_status()
return self.prepare_model(attrs=response.json())
def list(self, **kwargs) -> list[Secret]:
"""Report on Secrets.
Keyword Args:
filters (dict[str, Any]): Ignored.
Raises:
APIError: when error returned by service
"""
response = self.client.get("/secrets/json")
response.raise_for_status()
return [self.prepare_model(attrs=item) for item in response.json()]
def create(
self,
name: str,
data: bytes,
labels: Optional[Mapping[str, Any]] = None, # pylint: disable=unused-argument
driver: Optional[str] = None,
) -> Secret:
"""Create a Secret.
Args:
name: User-defined name of the secret.
data: Secret to be registered with Podman service.
labels: Ignored.
driver: Secret driver.
Raises:
APIError: when service returns an error
"""
params = {
"name": name,
"driver": driver,
}
response = self.client.post("/secrets/create", params=params, data=data)
response.raise_for_status()
body = response.json()
return self.get(body["ID"])
def remove(
self,
secret_id: Union[Secret, str],
all: Optional[bool] = None, # pylint: disable=redefined-builtin
):
"""Delete secret.
Podman only
Args:
secret_id: Identifier of Secret to delete.
all: When True, delete all secrets.
Raises:
NotFound: when Secret does not exist
APIError: when an error returned by service
"""
if isinstance(secret_id, Secret):
secret_id = secret_id.id
response = self.client.delete(f"/secrets/{secret_id}", params={"all": all})
response.raise_for_status()
|
