/
__init__.py
88 lines (66 loc) · 2.23 KB
/
__init__.py
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
import argparse
import json
class SecretNotFound(Exception):
pass
class Secrets:
def __init__(self, project):
self.project = project
self.client = None
def set(self, name, value):
return self._set_secret(name, value)
def get(self, name):
return self._get_secret(name)
def get_or_none(self, name):
try:
return self.get(name)
except SecretNotFound:
return None
def list(self):
raise NotImplementedError()
def delete(self, name):
return self._delete_secret(name)
def _set_secret(self, name, value):
client = self._client()
try:
self._get_secret(name)
except SecretNotFound:
parent = f"projects/{self.project}"
client.create_secret(
request={
"parent": parent,
"secret_id": name,
"secret": {"replication": {"automatic": {}}},
}
)
parent = client.secret_path(self.project, name)
secret_bytes = value.encode("utf-8")
secret_parent = client.secret_path(self.project, name)
return client.add_secret_version(
request={"parent": secret_parent, "payload": {"data": secret_bytes}}
)
def _get_secret(self, name):
from google.api_core import exceptions
client = self._client()
try:
response = client.access_secret_version(
request={
"name": f"projects/{self.project}/secrets/{name}/versions/latest"
}
)
return response.payload.data.decode("utf-8")
except (exceptions.NotFound, exceptions.PermissionDenied):
raise SecretNotFound()
def _delete_secret(self, name):
try:
self._get_secret(name)
except SecretNotFound:
return
client = self._client()
name = client.secret_path(self.project, name)
client.delete_secret(name)
def _client(self):
if self.client:
return self.client
from google.cloud import secretmanager_v1
self.client = secretmanager_v1.SecretManagerServiceClient()
return self.client