forked from googleapis/google-cloud-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathkey.py
More file actions
274 lines (214 loc) · 8.54 KB
/
key.py
File metadata and controls
274 lines (214 loc) · 8.54 KB
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
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
# Copyright 2014 Google Inc. 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.
"""Create / interact with gcloud datastore keys."""
import copy
from itertools import izip
import six
from gcloud.datastore import datastore_v1_pb2 as datastore_pb
class Key(object):
"""An immutable representation of a datastore Key.
To create a basic key:
>>> Key('EntityKind', 1234)
<Key[{'kind': 'EntityKind', 'id': 1234}]>
>>> Key('EntityKind', 'foo')
<Key[{'kind': 'EntityKind', 'name': 'foo'}]>
To create a key with a parent:
>>> Key('Parent', 'foo', 'Child', 1234)
<Key[{'kind': 'Parent', 'name': 'foo'}, {'kind': 'Child', 'id': 1234}]>
To create a paritial key:
>>> Key('Parent', 'foo', 'Child')
<Key[{'kind': 'Parent', 'name': 'foo'}, {'kind': 'Child'}]>
.. automethod:: __init__
"""
def __init__(self, *path_args, **kwargs):
"""Constructor / initializer for a key.
:type path_args: tuple of strings and ints
:param path_args: May represent a partial (odd length) or full (even
length) key path.
:type namespace: :class:`str`
:param namespace: A namespace identifier for the key. Can only be
passed as a keyword argument.
:type dataset_id: string
:param dataset_id: The dataset ID associated with the key. Can only be
passed as a keyword argument.
# This note will be obsolete by the end of #451.
.. note::
The key's ``_dataset_id`` field must be None for keys created
by application code. The
:func:`gcloud.datastore.helpers.key_from_protobuf` factory
will be set the field to an appropriate value for keys
returned from the datastore backend. The application
**must** treat any value set by the back-end as opaque.
"""
self._path = self._parse_path(path_args)
self._flat_path = path_args
self._parent = None
self._namespace = kwargs.get('namespace')
self._dataset_id = kwargs.get('dataset_id')
@staticmethod
def _parse_path(path_args):
"""Parses positional arguments into key path with kinds and IDs.
:rtype: list of dict
:returns: A list of key parts with kind and id or name set.
:raises: `ValueError` if there are no `path_args`, if one of the
kinds is not a string or if one of the IDs/names is not
a string or an integer.
"""
if len(path_args) == 0:
raise ValueError('Key path must not be empty.')
kind_list = path_args[::2]
id_or_name_list = path_args[1::2]
# Dummy sentinel value to pad incomplete key to even length path.
partial_ending = object()
if len(path_args) % 2 == 1:
id_or_name_list += (partial_ending,)
result = []
for kind, id_or_name in izip(kind_list, id_or_name_list):
curr_key_part = {}
if isinstance(kind, six.string_types):
curr_key_part['kind'] = kind
else:
raise ValueError(kind, 'Kind was not a string.')
if isinstance(id_or_name, six.string_types):
curr_key_part['name'] = id_or_name
elif isinstance(id_or_name, six.integer_types):
curr_key_part['id'] = id_or_name
elif id_or_name is not partial_ending:
raise ValueError(id_or_name,
'ID/name was not a string or integer.')
result.append(curr_key_part)
return result
def _clone(self):
"""Duplicates the Key.
We make a shallow copy of the :class:`gcloud.datastore.dataset.Dataset`
because it holds a reference an authenticated connection,
which we don't want to lose.
:rtype: :class:`gcloud.datastore.key.Key`
:returns: a new `Key` instance
"""
return copy.deepcopy(self)
def to_protobuf(self):
"""Return a protobuf corresponding to the key.
:rtype: :class:`gcloud.datastore.datastore_v1_pb2.Key`
:returns: The Protobuf representing the key.
"""
key = datastore_pb.Key()
if self.dataset_id is not None:
key.partition_id.dataset_id = self.dataset_id
if self.namespace:
key.partition_id.namespace = self.namespace
for item in self.path:
element = key.path_element.add()
if 'kind' in item:
element.kind = item['kind']
if 'id' in item:
element.id = item['id']
if 'name' in item:
element.name = item['name']
return key
@property
def is_partial(self):
"""Boolean indicating if the key has an ID (or name).
:rtype: :class:`bool`
:returns: True if the last element of the key's path does not have
an 'id' or a 'name'.
"""
return self.id_or_name is None
@property
def namespace(self):
"""Namespace getter.
:rtype: :class:`str`
:returns: The namespace of the current key.
"""
return self._namespace
@property
def path(self):
"""Path getter.
Returns a copy so that the key remains immutable.
:rtype: :class:`str`
:returns: The (key) path of the current key.
"""
return copy.deepcopy(self._path)
@property
def flat_path(self):
"""Getter for the key path as a tuple.
:rtype: :class:`tuple` of string and int
:returns: The tuple of elements in the path.
"""
return self._flat_path
@property
def kind(self):
"""Kind getter. Based on the last element of path.
:rtype: :class:`str`
:returns: The kind of the current key.
"""
return self.path[-1]['kind']
@property
def id(self):
"""ID getter. Based on the last element of path.
:rtype: :class:`int`
:returns: The (integer) ID of the key.
"""
return self.path[-1].get('id')
@property
def name(self):
"""Name getter. Based on the last element of path.
:rtype: :class:`str`
:returns: The (string) name of the key.
"""
return self.path[-1].get('name')
@property
def id_or_name(self):
"""Getter. Based on the last element of path.
:rtype: :class:`int` (if 'id') or :class:`str` (if 'name')
:returns: The last element of the key's path if it is either an 'id'
or a 'name'.
"""
return self.id or self.name
@property
def dataset_id(self):
"""Dataset ID getter.
:rtype: :class:`str`
:returns: The key's dataset.
"""
return self._dataset_id
def _make_parent(self):
"""Creates a parent key for the current path.
Extracts all but the last element in the key path and creates a new
key, while still matching the namespace and the dataset ID.
:rtype: :class:`gcloud.datastore.key.Key` or `NoneType`
:returns: a new `Key` instance, whose path consists of all but the last
element of self's path. If self has only one path element,
returns None.
"""
if self.is_partial:
parent_args = self.flat_path[:-1]
else:
parent_args = self.flat_path[:-2]
if parent_args:
return Key(*parent_args, dataset_id=self.dataset_id,
namespace=self.namespace)
@property
def parent(self):
"""The parent of the current key.
:rtype: :class:`gcloud.datastore.key.Key` or `NoneType`
:returns: a new `Key` instance, whose path consists of all but the last
element of self's path. If self has only one path element,
returns None.
"""
if self._parent is None:
self._parent = self._make_parent()
return self._parent
def __repr__(self):
return '<Key%s>' % self.path