File: share_snapshots.py

package info (click to toggle)
python-manilaclient 5.4.0-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 3,768 kB
  • sloc: python: 49,541; makefile: 99; sh: 2
file content (254 lines) | stat: -rw-r--r-- 9,276 bytes parent folder | download
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
# Copyright 2012 NetApp
# All Rights Reserved.
#
#    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.
"""Interface for shares extension."""

from manilaclient import api_versions
from manilaclient import base
from manilaclient.common import constants


class ShareSnapshot(base.MetadataCapableResource):

    """Represent a snapshot of a share."""

    def __repr__(self):
        return "<ShareSnapshot: %s>" % self.id

    def update(self, **kwargs):
        """Update this snapshot."""
        self.manager.update(self, **kwargs)

    def reset_state(self, state):
        """Update the snapshot with the provided state."""
        self.manager.reset_state(self, state)

    def delete(self):
        """Delete this snapshot."""
        self.manager.delete(self)

    def force_delete(self):
        """Delete the specified snapshot ignoring its current state."""
        self.manager.force_delete(self)

    def unmanage_snapshot(self):
        """Unmanage this snapshot."""
        self.manager.unmanage(self)

    def allow(self, access_type, access_to):
        """Allow access to a share snapshot."""
        return self.manager.allow(self, access_type, access_to)

    def deny(self, id):
        """Denies access to a share snapshot."""
        return self.manager.deny(self, id)

    def access_list(self):
        return self.manager.access_list(self)


class ShareSnapshotManager(base.MetadataCapableManager):
    """Manage :class:`ShareSnapshot` resources."""
    resource_class = ShareSnapshot
    resource_path = '/snapshots'

    def _do_create(self, share, force=False, name=None, description=None,
                   metadata=None):
        """Create a snapshot of the given share.

        :param share_id: The ID of the share to snapshot.
        :param force: If force is True, create a snapshot even if the
                      share is busy. Default is False.
        :param name: Name of the snapshot
        :param description: Description of the snapshot
        :param metadata: dict - optional metadata to set on share creation
        :rtype: :class:`ShareSnapshot`
        """

        metadata = metadata if metadata is not None else dict()
        body = {'snapshot': {'share_id': base.getid(share),
                             'force': force,
                             'name': name,
                             'description': description,
                             'metadata': metadata}}
        return self._create('/snapshots', body, 'snapshot')

    @api_versions.wraps("2.0", "2.72")
    def create(self, share, force=False, name=None, description=None):
        return self._do_create(share, force, name, description)

    @api_versions.wraps("2.73")
    def create(self, share, force=False, name=None, description=None,# noqa F811
               metadata=None):
        return self._do_create(share, force, name, description, metadata)

    @api_versions.wraps("2.12")
    def manage(self, share, provider_location,
               driver_options=None,
               name=None, description=None):
        """Manage an existing share snapshot.

        :param share: The share object.
        :param provider_location: The provider location of
                                  the snapshot on the backend.
        :param driver_options: dict - custom set of key-values.
        :param name: text - name of new snapshot
        :param description: - description for new snapshot
        """
        driver_options = driver_options if driver_options else {}
        body = {
            'share_id': base.getid(share),
            'provider_location': provider_location,
            'driver_options': driver_options,
            'name': name,
            'description': description,
        }
        return self._create('/snapshots/manage', {'snapshot': body},
                            'snapshot')

    @api_versions.wraps("2.12")
    def unmanage(self, snapshot):
        """Unmanage a share snapshot.

        :param snapshot: either snapshot object or text with its ID.
        """
        return self._action("unmanage", snapshot)

    def get(self, snapshot):
        """Get a snapshot.

        :param snapshot: The :class:`ShareSnapshot` instance or string with ID
            of snapshot to delete.
        :rtype: :class:`ShareSnapshot`
        """
        snapshot_id = base.getid(snapshot)
        return self._get('/snapshots/%s' % snapshot_id, 'snapshot')

    def list(self, detailed=True, search_opts=None, sort_key=None,
             sort_dir=None):
        """Get a list of snapshots of shares.

        :param search_opts: Search options to filter out shares.
        :param sort_key: Key to be sorted.
        :param sort_dir: Sort direction, should be 'desc' or 'asc'.
        :rtype: list of :class:`ShareSnapshot`
        """
        search_opts = search_opts or {}

        if sort_key is not None:
            if sort_key in constants.SNAPSHOT_SORT_KEY_VALUES:
                search_opts['sort_key'] = sort_key
            else:
                raise ValueError(
                    'sort_key must be one of the following: %s.'
                    % ', '.join(constants.SNAPSHOT_SORT_KEY_VALUES))

        if sort_dir is not None:
            if sort_dir in constants.SORT_DIR_VALUES:
                search_opts['sort_dir'] = sort_dir
            else:
                raise ValueError(
                    'sort_dir must be one of the following: %s.'
                    % ', '.join(constants.SORT_DIR_VALUES))

        query_string = self._build_query_string(search_opts)

        if detailed:
            path = "/snapshots/detail%s" % (query_string,)
        else:
            path = "/snapshots%s" % (query_string,)

        return self._list(path, 'snapshots')

    def delete(self, snapshot):
        """Delete a snapshot of a share.

        :param snapshot: The :class:`ShareSnapshot` to delete.
        """
        self._delete("/snapshots/%s" % base.getid(snapshot))

    def _do_force_delete(self, snapshot, action_name="force_delete"):
        """Delete the specified snapshot ignoring its current state."""
        return self._action(action_name, base.getid(snapshot))

    @api_versions.wraps("1.0", "2.6")
    def force_delete(self, snapshot):
        return self._do_force_delete(snapshot, "os-force_delete")

    @api_versions.wraps("2.7")  # noqa
    def force_delete(self, snapshot):  # noqa
        return self._do_force_delete(snapshot, "force_delete")

    def update(self, snapshot, **kwargs):
        """Update a snapshot.

        :param snapshot: The :class:`ShareSnapshot` instance or string with ID
            of snapshot to delete.
        :rtype: :class:`ShareSnapshot`
        """
        if not kwargs:
            return

        body = {'snapshot': kwargs, }
        snapshot_id = base.getid(snapshot)
        return self._update("/snapshots/%s" % snapshot_id, body)

    def _do_reset_state(self, snapshot, state, action_name="reset_status"):
        """Update the specified share snapshot with the provided state."""
        return self._action(action_name, snapshot, {"status": state})

    @api_versions.wraps("1.0", "2.6")
    def reset_state(self, snapshot, state):
        return self._do_reset_state(snapshot, state, "os-reset_status")

    @api_versions.wraps("2.7")  # noqa
    def reset_state(self, snapshot, state):  # noqa
        return self._do_reset_state(snapshot, state, "reset_status")

    def _do_allow(self, snapshot, access_type, access_to):
        access_params = {
            'access_type': access_type,
            'access_to': access_to,
        }

        return self._action('allow_access', snapshot,
                            access_params)[1]['snapshot_access']

    @api_versions.wraps("2.32")
    def allow(self, snapshot, access_type, access_to):
        return self._do_allow(snapshot, access_type, access_to)

    def _do_deny(self, snapshot, id):
        return self._action('deny_access', snapshot, {'access_id': id})

    @api_versions.wraps("2.32")
    def deny(self, snapshot, id):
        return self._do_deny(snapshot, id)

    def _do_access_list(self, snapshot):
        snapshot_id = base.getid(snapshot)
        access_list = self._list("/snapshots/%s/access-list" % snapshot_id,
                                 'snapshot_access_list')
        return access_list

    @api_versions.wraps("2.32")
    def access_list(self, snapshot):
        return self._do_access_list(snapshot)

    def _action(self, action, snapshot, info=None, **kwargs):
        """Perform a snapshot 'action'."""
        body = {action: info}
        self.run_hooks('modify_body_for_action', body, **kwargs)
        url = '/snapshots/%s/action' % base.getid(snapshot)
        return self.api.client.post(url, body=body)