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
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import typing as ty
import typing_extensions as ty_ext
from openstack import exceptions
from openstack import resource
from openstack import utils
# https://github.com/python/mypy/issues/11583
class _TagQueryParameters(ty.TypedDict):
tags: str
any_tags: str
not_tags: str
not_any_tags: str
class TagMixin(resource.ResourceMixinProtocol):
_tag_query_parameters: _TagQueryParameters = {
'tags': 'tags',
'any_tags': 'tags-any',
'not_tags': 'not-tags',
'not_any_tags': 'not-tags-any',
}
#: A list of associated tags
#: *Type: list of tag strings*
tags = resource.Body('tags', type=list, default=[])
def fetch_tags(self, session: resource.AdapterT) -> ty_ext.Self:
"""Lists tags set on the entity.
:param session: The session to use for making this request.
:return: The list with tags attached to the entity
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.get(url)
exceptions.raise_from_response(response)
# NOTE(gtema): since this is a common method
# we can't rely on the resource_key, because tags are returned
# without resource_key. Do parse response here
json = response.json()
if 'tags' in json:
self._body.attributes.update({'tags': json['tags']})
return self
def set_tags(
self, session: resource.AdapterT, tags: list[str]
) -> ty_ext.Self:
"""Sets/Replaces all tags on the resource.
:param session: The session to use for making this request.
:param list tags: List with tags to be set on the resource
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.put(url, json={'tags': tags})
exceptions.raise_from_response(response)
self._body.attributes.update({'tags': tags})
return self
def remove_all_tags(self, session: resource.AdapterT) -> ty_ext.Self:
"""Removes all tags on the entity.
:param session: The session to use for making this request.
"""
url = utils.urljoin(self.base_path, self.id, 'tags')
session = self._get_session(session)
response = session.delete(url)
exceptions.raise_from_response(response)
self._body.attributes.update({'tags': []})
return self
def check_tag(self, session: resource.AdapterT, tag: str) -> ty_ext.Self:
"""Checks if tag exists on the entity.
If the tag does not exist a 404 will be returned
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.get(url)
exceptions.raise_from_response(
response, error_message='Tag does not exist'
)
return self
def add_tag(self, session: resource.AdapterT, tag: str) -> ty_ext.Self:
"""Adds a single tag to the resource.
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.put(url)
exceptions.raise_from_response(response)
# we do not want to update tags directly
tags = self.tags
tags.append(tag)
self._body.attributes.update({'tags': tags})
return self
def remove_tag(self, session: resource.AdapterT, tag: str) -> ty_ext.Self:
"""Removes a single tag from the specified resource.
:param session: The session to use for making this request.
:param tag: The tag as a string.
"""
url = utils.urljoin(self.base_path, self.id, 'tags', tag)
session = self._get_session(session)
response = session.delete(url)
exceptions.raise_from_response(response)
# we do not want to update tags directly
tags = self.tags
try:
# NOTE(gtema): if tags were not fetched, but request suceeded
# it is ok. Just ensure tag does not exist locally
tags.remove(tag)
except ValueError:
pass # do nothing!
self._body.attributes.update({'tags': tags})
return self
|
