py_tasks_melange: comparison thirdparty/google_appengine/google/appengine/api/datastore_file

equal deleted inserted replaced

-:27971a13089f
+:2e0b0af889be
 _MAX_QUERY_OFFSET = 1000
 _MAX_QUERY_COMPONENTS = 100
+_BATCH_SIZE = 20
 class _StoredEntity(object):
 """Simple wrapper around an entity stored by the stub.
 Public properties:
 Public properties:
 cursor: the integer cursor
 count: the original total number of results
 keys_only: whether the query is keys_only
+Class attributes:
+_next_cursor: the next cursor to allocate
+_next_cursor_lock: protects _next_cursor
 """
+_next_cursor = 1
+_next_cursor_lock = threading.Lock()
 def __init__(self, results, keys_only):
 """Constructor.
 Args:
 # the query results, in order, such that pop(0) is the next result
 keys_only: integer
 """
 self.__results = results
 self.count = len(results)
 self.keys_only = keys_only
-self.cursor = id(self)
+self._next_cursor_lock.acquire()
+try:
+self.cursor = _Cursor._next_cursor
+_Cursor._next_cursor += 1
+finally:
+self._next_cursor_lock.release()
 def PopulateQueryResult(self, result, count):
 """Populates a QueryResult with this cursor and the given number of results.
 Args:
 """
 if not self.__trusted and app_id != self.__app_id:
 raise datastore_errors.BadRequestError(
 'app %s cannot access app %s\'s data' % (self.__app_id, app_id))
+def __ValidateKey(self, key):
-def _AppKindForKey(self, key):
+"""Validate this key.
+Args:
+key: entity_pb.Reference
+Raises:
+datastore_errors.BadRequestError: if the key is invalid
+"""
+assert isinstance(key, entity_pb.Reference)
+self.__ValidateAppId(key.app())
+for elem in key.path().element_list():
+if elem.has_id() == elem.has_name():
+raise datastore_errors.BadRequestError(
+'each key path element should have id or name but not both: %r' % key)
+def _AppIdNamespaceKindForKey(self, key):
 """ Get (app, kind) tuple from given key.
 The (app, kind) tuple is used as an index into several internal
 dictionaries, e.g. __entities.
 Args:
 entity: entity_pb.EntityProto
 """
 key = entity.key()
-app_kind = self._AppKindForKey(key)
+app_kind = self._AppIdNamespaceKindForKey(key)
 if app_kind not in self.__entities:
 self.__entities[app_kind] = {}
 self.__entities[app_kind][key] = _StoredEntity(entity)
 if app_kind in self.__schema_cache:
 os.rename(tmpfile.name, filename)
 finally:
 self.__file_lock.release()
 def MakeSyncCall(self, service, call, request, response):
-""" The main RPC entry point. service must be 'datastore_v3'. So far, the
+""" The main RPC entry point. service must be 'datastore_v3'.
-supported calls are 'Get', 'Put', 'RunQuery', 'Next', and 'Count'.
+"""
-"""
+self.assertPbIsInitialized(request)
 super(DatastoreFileStub, self).MakeSyncCall(service,
 call,
 request,
 response)
+self.assertPbIsInitialized(response)
+def assertPbIsInitialized(self, pb):
+"""Raises an exception if the given PB is not initialized and valid."""
 explanation = []
-assert response.IsInitialized(explanation), explanation
+assert pb.IsInitialized(explanation), explanation
+pb.Encode()
 def QueryHistory(self):
 """Returns a dict that maps Query PBs to times they've been run.
 """
 return dict((pb, times) for pb, times in self.__query_history.items()
 if pb.app() == self.__app_id)
 def _Dynamic_Put(self, put_request, put_response):
 clones = []
 for entity in put_request.entity_list():
-self.__ValidateAppId(entity.key().app())
+self.__ValidateKey(entity.key())
 clone = entity_pb.EntityProto()
 clone.CopyFrom(entity)
 for property in clone.property_list():
 else:
 entities = self.__entities
 for key in get_request.key_list():
 self.__ValidateAppId(key.app())
-app_kind = self._AppKindForKey(key)
+app_kind = self._AppIdNamespaceKindForKey(key)
 group = get_response.add_entity()
 try:
 entity = entities[app_kind][key].protobuf
 except KeyError:
 def _Dynamic_Delete(self, delete_request, delete_response):
 self.__entities_lock.acquire()
 try:
 for key in delete_request.key_list():
 self.__ValidateAppId(key.app())
-app_kind = self._AppKindForKey(key)
+app_kind = self._AppIdNamespaceKindForKey(key)
 try:
 del self.__entities[app_kind][key]
 if not self.__entities[app_kind]:
 del self.__entities[app_kind]
 entities = self.__tx_snapshot
 else:
 entities = self.__entities
 self.__tx_lock.release()
-app = query.app()
+app_id_namespace = datastore_types.parse_app_id_namespace(query.app())
-self.__ValidateAppId(app)
+app_id = app_id_namespace.app_id()
+self.__ValidateAppId(app_id)
 if query.has_offset() and query.offset() > _MAX_QUERY_OFFSET:
 raise apiproxy_errors.ApplicationError(
 datastore_pb.Error.BAD_REQUEST, 'Too big query offset.')
 raise apiproxy_errors.ApplicationError(
 datastore_pb.Error.BAD_REQUEST,
 ('query is too large. may not have more than %s filters'
 ' + sort orders ancestor total' % _MAX_QUERY_COMPONENTS))
+(filters, orders) = datastore_index.Normalize(query.filter_list(),
+query.order_list())
 if self.__require_indexes:
 required, kind, ancestor, props, num_eq_filters = datastore_index.CompositeIndexForQuery(query)
 if required:
 required_key = kind, ancestor, props
-indexes = self.__indexes.get(app)
+indexes = self.__indexes.get(app_id)
 if not indexes:
 raise apiproxy_errors.ApplicationError(
 datastore_pb.Error.NEED_INDEX,
 "This query requires a composite index, but none are defined. "
 "You must create an index.yaml file in your application root.")
 datastore_pb.Error.NEED_INDEX,
 "This query requires a composite index that is not defined. "
 "You must update the index.yaml file in your application root.")
 try:
-query.set_app(app)
+query.set_app(app_id_namespace.to_encoded())
-results = entities[app, query.kind()].values()
+if query.has_kind():
-results = [entity.native for entity in results]
+results = entities[app_id_namespace.to_encoded(), query.kind()].values()
+results = [entity.native for entity in results]
+else:
+results = []
+for key in entities:
+if key[0] == app_id_namespace.to_encoded():
+results += [entity.native for entity in entities[key].values()]
 except KeyError:
 results = []
 if query.has_ancestor():
 ancestor_path = query.ancestor().path().element_list()
 for value in values:
 if type(value) not in datastore_types._RAW_PROPERTY_TYPES:
 return True
 return False
-for filt in query.filter_list():
+for filt in filters:
 assert filt.op() != datastore_pb.Query_Filter.IN
 prop = filt.property(0).name().decode('utf-8')
 op = operators[filt.op()]
 return False
 results = filter(passes_filter, results)
-for order in query.order_list():
+for order in orders:
 prop = order.property().decode('utf-8')
 results = [entity for entity in results if has_prop_indexed(entity, prop)]
 def order_compare_entities(a, b):
 """ Return a negative, zero or positive number depending on whether
 entity a is considered smaller than, equal to, or larger than b,
 according to the query's orderings. """
 cmped = 0
-for o in query.order_list():
+for o in orders:
 prop = o.property().decode('utf-8')
 reverse = (o.direction() is datastore_pb.Query_Order.DESCENDING)
 a_val = datastore._GetPropertyValue(a, prop)
 self.__query_history[clone] = 1
 self.__WriteHistory()
 cursor = _Cursor(results, query.keys_only())
 self.__queries[cursor.cursor] = cursor
-cursor.PopulateQueryResult(query_result, 0)
+if query.has_count():
+count = query.count()
+elif query.has_limit():
+count = query.limit()
+else:
+count = _BATCH_SIZE
+cursor.PopulateQueryResult(query_result, count)
 def _Dynamic_Next(self, next_request, query_result):
 cursor_handle = next_request.cursor().cursor()
 try:
 cursor = self.__queries[cursor_handle]
 except KeyError:
 raise apiproxy_errors.ApplicationError(
 datastore_pb.Error.BAD_REQUEST, 'Cursor %d not found' % cursor_handle)
-cursor.PopulateQueryResult(query_result, next_request.count())
+count = _BATCH_SIZE
+if next_request.has_count():
+count = next_request.count()
+cursor.PopulateQueryResult(query_result, count)
 def _Dynamic_Count(self, query, integer64proto):
 self.__ValidateAppId(query.app())
 query_result = datastore_pb.QueryResult()
 self._Dynamic_RunQuery(query, query_result)
 self.__entities = self.__tx_snapshot
 self.__tx_snapshot = {}
 self.__tx_lock.release()
-def _Dynamic_GetSchema(self, app_str, schema):
+def _Dynamic_GetSchema(self, req, schema):
-minint = -sys.maxint - 1
+app_str = req.app()
-try:
-minfloat = float('-inf')
-except ValueError:
-minfloat = -1e300000
-app_str = app_str.value()
 self.__ValidateAppId(app_str)
 kinds = []
 for app, kind in self.__entities:
-if app == app_str:
+if (app != app_str or
-app_kind = (app, kind)
+(req.has_start_kind() and kind < req.start_kind()) or
-if app_kind in self.__schema_cache:
+(req.has_end_kind() and kind > req.end_kind())):
-kinds.append(self.__schema_cache[app_kind])
+continue
-continue
+app_kind = (app, kind)
-kind_pb = entity_pb.EntityProto()
+if app_kind in self.__schema_cache:
-kind_pb.mutable_key().set_app('')
+kinds.append(self.__schema_cache[app_kind])
-kind_pb.mutable_key().mutable_path().add_element().set_type(kind)
+continue
-kind_pb.mutable_entity_group()
+kind_pb = entity_pb.EntityProto()
-props = {}
+kind_pb.mutable_key().set_app('')
+kind_pb.mutable_key().mutable_path().add_element().set_type(kind)
-for entity in self.__entities[app_kind].values():
+kind_pb.mutable_entity_group()
-for prop in entity.protobuf.property_list():
-if prop.name() not in props:
+props = {}
-props[prop.name()] = entity_pb.PropertyValue()
-props[prop.name()].MergeFrom(prop.value())
+for entity in self.__entities[app_kind].values():
+for prop in entity.protobuf.property_list():
-for value_pb in props.values():
+if prop.name() not in props:
-if value_pb.has_int64value():
+props[prop.name()] = entity_pb.PropertyValue()
-value_pb.set_int64value(minint)
+props[prop.name()].MergeFrom(prop.value())
-if value_pb.has_booleanvalue():
-value_pb.set_booleanvalue(False)
+for value_pb in props.values():
-if value_pb.has_stringvalue():
+if value_pb.has_int64value():
-value_pb.set_stringvalue('')
+value_pb.set_int64value(0)
-if value_pb.has_doublevalue():
+if value_pb.has_booleanvalue():
-value_pb.set_doublevalue(minfloat)
+value_pb.set_booleanvalue(False)
-if value_pb.has_pointvalue():
+if value_pb.has_stringvalue():
-value_pb.mutable_pointvalue().set_x(minfloat)
+value_pb.set_stringvalue('none')
-value_pb.mutable_pointvalue().set_y(minfloat)
+if value_pb.has_doublevalue():
-if value_pb.has_uservalue():
+value_pb.set_doublevalue(0.0)
-value_pb.mutable_uservalue().set_gaiaid(minint)
+if value_pb.has_pointvalue():
-value_pb.mutable_uservalue().set_email('')
+value_pb.mutable_pointvalue().set_x(0.0)
-value_pb.mutable_uservalue().set_auth_domain('')
+value_pb.mutable_pointvalue().set_y(0.0)
-value_pb.mutable_uservalue().clear_nickname()
+if value_pb.has_uservalue():
-elif value_pb.has_referencevalue():
+value_pb.mutable_uservalue().set_gaiaid(0)
-value_pb.clear_referencevalue()
+value_pb.mutable_uservalue().set_email('none')
-value_pb.mutable_referencevalue().set_app('')
+value_pb.mutable_uservalue().set_auth_domain('none')
+value_pb.mutable_uservalue().clear_nickname()
-for name, value_pb in props.items():
+value_pb.mutable_uservalue().clear_obfuscated_gaiaid()
-prop_pb = kind_pb.add_property()
+if value_pb.has_referencevalue():
-prop_pb.set_name(name)
+value_pb.clear_referencevalue()
-prop_pb.set_multiple(False)
+value_pb.mutable_referencevalue().set_app('none')
-prop_pb.mutable_value().CopyFrom(value_pb)
+pathelem = value_pb.mutable_referencevalue().add_pathelement()
+pathelem.set_type('none')
-kinds.append(kind_pb)
+pathelem.set_name('none')
-self.__schema_cache[app_kind] = kind_pb
+for name, value_pb in props.items():
+prop_pb = kind_pb.add_property()
+prop_pb.set_name(name)
+prop_pb.set_multiple(False)
+prop_pb.mutable_value().CopyFrom(value_pb)
+kinds.append(kind_pb)
+self.__schema_cache[app_kind] = kind_pb
 for kind_pb in kinds:
-schema.add_kind().CopyFrom(kind_pb)
+kind = schema.add_kind()
+kind.CopyFrom(kind_pb)
+if not req.properties():
+kind.clear_property()
+schema.set_more_results(False)
+def _Dynamic_AllocateIds(self, allocate_ids_request, allocate_ids_response):
+model_key = allocate_ids_request.model_key()
+size = allocate_ids_request.size()
+self.__ValidateAppId(model_key.app())
+try:
+self.__id_lock.acquire()
+start = self.__next_id
+self.__next_id += size
+end = self.__next_id - 1
+finally:
+self.__id_lock.release()
+allocate_ids_response.set_start(start)
+allocate_ids_response.set_end(end)
 def _Dynamic_CreateIndex(self, index, id_response):
 self.__ValidateAppId(index.app_id())
 if index.id() != 0:
 raise apiproxy_errors.ApplicationError(datastore_pb.Error.BAD_REQUEST,

changeset 2864	2e0b0af889be
parent 2413	d0b7dac5325c
child 3031	7678f72140e6