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

equal deleted inserted replaced

-:a440ced9a75f
+:df109be0567c
 import datetime
 import logging
 import os
-import pickle
 import struct
 import sys
 import tempfile
 import threading
 import warnings
+import cPickle as pickle
 from google.appengine.api import api_base_pb
+from google.appengine.api import apiproxy_stub
 from google.appengine.api import datastore
 from google.appengine.api import datastore_admin
 from google.appengine.api import datastore_errors
 from google.appengine.api import datastore_types
 from google.appengine.api import users
 entity_pb.Reference.__hash__ = lambda self: hash(self.Encode())
 datastore_pb.Query.__hash__ = lambda self: hash(self.Encode())
+_MAXIMUM_RESULTS = 1000
+_MAX_QUERY_OFFSET = 4000
 class _StoredEntity(object):
 """Simple wrapper around an entity stored by the stub.
 Public properties:
-native: Native protobuf Python object, entity_pb.EntityProto.
+protobuf: Native protobuf Python object, entity_pb.EntityProto.
-encoded: Encoded binary representation of above protobuf.
+encoded_protobuf: Encoded binary representation of above protobuf.
+native: datastore.Entity instance.
 """
 def __init__(self, entity):
 """Create a _StoredEntity object and store an entity.
 Args:
 entity: entity_pb.EntityProto to store.
 """
-self.native = entity
+self.protobuf = entity
-self.encoded = entity.Encode()
+self.encoded_protobuf = entity.Encode()
+self.native = datastore.Entity._FromPb(entity)
-class DatastoreFileStub(object):
+class DatastoreFileStub(apiproxy_stub.APIProxyStub):
 """ Persistent stub for the Python datastore API.
 Stores all entities in memory, and persists them to a file as pickled
 protocol buffers. A DatastoreFileStub instance handles a single app's data
 and is backed by files on disk.
 type(None): 0,
 unicode: entity_pb.PropertyValue.kstringValue,
 users.User: entity_pb.PropertyValue.kUserValueGroup,
 }
-def __init__(self, app_id, datastore_file, history_file,
+def __init__(self,
-require_indexes=False):
+app_id,
+datastore_file,
+history_file,
+require_indexes=False,
+service_name='datastore_v3'):
 """Constructor.
 Initializes and loads the datastore from the backing files, if they exist.
 Args:
 not to use a file.
 history_file: string, stores query history.  Use None as with
 datastore_file.
 require_indexes: bool, default False.  If True, composite indexes must
 exist in index.yaml for queries that need them.
-"""
+service_name: Service name expected for all calls.
+"""
+super(DatastoreFileStub, self).__init__(service_name)
 assert isinstance(app_id, basestring) and app_id != ''
 self.__app_id = app_id
 self.__datastore_file = datastore_file
 self.__history_file = history_file
 self.__entities = {}
+self.__schema_cache = {}
 self.__tx_snapshot = {}
 self.__queries = {}
 queries. """
 self.__entities = {}
 self.__queries = {}
 self.__transactions = {}
 self.__query_history = {}
+self.__schema_cache = {}
+def _AppKindForKey(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:
+key: entity_pb.Reference
+Returns:
+Tuple (app, kind), both are unicode strings.
+"""
+last_path = key.path().element_list()[-1]
+return key.app(), last_path.type()
+def _StoreEntity(self, entity):
+""" Store the given entity.
+Args:
+entity: entity_pb.EntityProto
+"""
+key = entity.key()
+app_kind = self._AppKindForKey(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:
+del self.__schema_cache[app_kind]
 def Read(self):
 """ Reads the datastore and history files into memory.
 The in-memory query history is cleared, but the datastore is *not*
 entity = entity_pb.EntityProto(encoded_entity)
 except pb_exceptions, e:
 raise datastore_errors.InternalError(error_msg %
 (self.__datastore_file, e))
+self._StoreEntity(entity)
 last_path = entity.key().path().element_list()[-1]
-app_kind = (entity.key().app(), last_path.type())
-kind_dict = self.__entities.setdefault(app_kind, {})
-kind_dict[entity.key()] = _StoredEntity(entity)
 if last_path.has_id() and last_path.id() >= self.__next_id:
 self.__next_id = last_path.id() + 1
 self.__query_history = {}
 for encoded_query, count in self.__ReadPickled(self.__history_file):
 """
 if self.__datastore_file and self.__datastore_file != '/dev/null':
 encoded = []
 for kind_dict in self.__entities.values():
 for entity in kind_dict.values():
-encoded.append(entity.encoded)
+encoded.append(entity.encoded_protobuf)
 self.__WritePickled(encoded, self.__datastore_file)
 def __WriteHistory(self):
 """ Writes out the history file. Be careful! If the file already exist,
 """
 if not filename or filename == '/dev/null' or not obj:
 return
 tmpfile = openfile(os.tempnam(os.path.dirname(filename)), 'wb')
-pickle.dump(obj, tmpfile, 1)
+pickler = pickle.Pickler(tmpfile, protocol=1)
+pickler.fast = True
+pickler.dump(obj)
 tmpfile.close()
 self.__file_lock.acquire()
 try:
 try:
 def MakeSyncCall(self, service, call, request, response):
 """ The main RPC entry point. service must be 'datastore_v3'. So far, the
 supported calls are 'Get', 'Put', 'RunQuery', 'Next', and 'Count'.
 """
+super(DatastoreFileStub, self).MakeSyncCall(service,
-assert service == 'datastore_v3'
+call,
+request,
+response)
 explanation = []
-assert request.IsInitialized(explanation), explanation
-(getattr(self, "_Dynamic_" + call))(request, response)
 assert response.IsInitialized(explanation), explanation
 def QueryHistory(self):
 """Returns a dict that maps Query PBs to times they've been run.
 """
 clones.append(clone)
 assert clone.has_key()
 assert clone.key().path().element_size() > 0
-app = clone.key().app()
 last_path = clone.key().path().element_list()[-1]
 if last_path.id() == 0 and not last_path.has_name():
 self.__id_lock.acquire()
 last_path.set_id(self.__next_id)
 self.__next_id += 1
 self.__entities_lock.acquire()
 try:
 for clone in clones:
-last_path = clone.key().path().element_list()[-1]
+self._StoreEntity(clone)
-kind_dict = self.__entities.setdefault((app, last_path.type()), {})
-kind_dict[clone.key()] = _StoredEntity(clone)
 finally:
 self.__entities_lock.release()
 if not put_request.has_transaction():
 self.__WriteDatastore()
 put_response.key_list().extend([c.key() for c in clones])
 def _Dynamic_Get(self, get_request, get_response):
 for key in get_request.key_list():
-app = key.app()
+app_kind = self._AppKindForKey(key)
-last_path = key.path().element_list()[-1]
 group = get_response.add_entity()
 try:
-entity = self.__entities[app, last_path.type()][key].native
+entity = self.__entities[app_kind][key].protobuf
 except KeyError:
 entity = None
 if entity:
 group.mutable_entity().CopyFrom(entity)
 def _Dynamic_Delete(self, delete_request, delete_response):
 self.__entities_lock.acquire()
 try:
 for key in delete_request.key_list():
+app_kind = self._AppKindForKey(key)
 try:
-app = key.app()
+del self.__entities[app_kind][key]
-kind = key.path().element_list()[-1].type()
+if not self.__entities[app_kind]:
-del self.__entities[app, kind][key]
+del self.__entities[app_kind]
-if not self.__entities[app, kind]:
-del self.__entities[app, kind]
+del self.__schema_cache[app_kind]
 except KeyError:
 pass
 if not delete_request.has_transaction():
 self.__WriteDatastore()
 raise apiproxy_errors.ApplicationError(
 datastore_pb.Error.BAD_REQUEST, "Can't query inside a transaction.")
 else:
 self.__tx_lock.release()
+if query.has_offset() and query.offset() > _MAX_QUERY_OFFSET:
+raise apiproxy_errors.ApplicationError(
+datastore_pb.Error.BAD_REQUEST, "Too big query offset.")
 app = query.app()
 if self.__require_indexes:
-required_index = datastore_index.CompositeIndexForQuery(query)
+required, kind, ancestor, props, num_eq_filters = datastore_index.CompositeIndexForQuery(query)
-if required_index is not None:
+if required:
-kind, ancestor, props, num_eq_filters = required_index
 required_key = kind, ancestor, props
 indexes = self.__indexes.get(app)
 if not indexes:
 raise apiproxy_errors.ApplicationError(
 datastore_pb.Error.NEED_INDEX,
 "You must update the index.yaml file in your application root.")
 try:
 query.set_app(app)
 results = self.__entities[app, query.kind()].values()
-results = [datastore.Entity._FromPb(entity.native) for entity in results]
+results = [entity.native for entity in results]
 except KeyError:
 results = []
 if query.has_ancestor():
 ancestor_path = query.ancestor().path().element_list()
 assert filt.op() != datastore_pb.Query_Filter.IN
 prop = filt.property(0).name().decode('utf-8')
 op = operators[filt.op()]
+filter_val_list = [datastore_types.FromPropertyPb(filter_prop)
+for filter_prop in filt.property_list()]
 def passes(entity):
 """ Returns True if the entity passes the filter, False otherwise. """
-entity_vals = entity.get(prop, [])
+if prop in datastore_types._SPECIAL_PROPERTIES:
+entity_vals = self.__GetSpecialPropertyValue(entity, prop)
+else:
+entity_vals = entity.get(prop, [])
 if not isinstance(entity_vals, list):
 entity_vals = [entity_vals]
-entity_property_list = [datastore_types.ToPropertyPb(prop, value)
+for fixed_entity_val in entity_vals:
-for value in entity_vals]
-for entity_prop in entity_property_list:
-fixed_entity_val = datastore_types.FromPropertyPb(entity_prop)
 if type(fixed_entity_val) in datastore_types._RAW_PROPERTY_TYPES:
 continue
-for filter_prop in filt.property_list():
+for filter_val in filter_val_list:
-filter_val = datastore_types.FromPropertyPb(filter_prop)
 fixed_entity_type = self._PROPERTY_TYPE_TAGS.get(
 fixed_entity_val.__class__)
 filter_type = self._PROPERTY_TYPE_TAGS.get(filter_val.__class__)
 if fixed_entity_type == filter_type:
 comp = u'%r %s %r' % (fixed_entity_val, op, filter_val)
 return False
 results = filter(passes, results)
 def has_prop_indexed(entity, prop):
-"""Returns True if prop is in the entity and is not a raw property."""
+"""Returns True if prop is in the entity and is not a raw property, or
+is a special property."""
+if prop in datastore_types._SPECIAL_PROPERTIES:
+return True
 values = entity.get(prop, [])
 if not isinstance(values, (tuple, list)):
 values = [values]
 for value in values:
 for o in query.order_list():
 prop = o.property().decode('utf-8')
 reverse = (o.direction() is datastore_pb.Query_Order.DESCENDING)
-a_val = a[prop]
+if prop in datastore_types._SPECIAL_PROPERTIES:
-if isinstance(a_val, list):
+a_val = self.__GetSpecialPropertyValue(a, prop)
-a_val = sorted(a_val, order_compare_properties, reverse=reverse)[0]
+b_val = self.__GetSpecialPropertyValue(b, prop)
+else:
-b_val = b[prop]
+a_val = a[prop]
-if isinstance(b_val, list):
+if isinstance(a_val, list):
-b_val = sorted(b_val, order_compare_properties, reverse=reverse)[0]
+a_val = sorted(a_val, order_compare_properties, reverse=reverse)[0]
+b_val = b[prop]
+if isinstance(b_val, list):
+b_val = sorted(b_val, order_compare_properties, reverse=reverse)[0]
 cmped = order_compare_properties(a_val, b_val)
 if o.direction() is datastore_pb.Query_Order.DESCENDING:
 cmped = -cmped
 limit = len(results)
 if query.has_offset():
 offset = query.offset()
 if query.has_limit():
 limit = query.limit()
+if limit > _MAXIMUM_RESULTS:
+limit = _MAXIMUM_RESULTS
 results = results[offset:limit + offset]
 clone = datastore_pb.Query()
 clone.CopyFrom(query)
 clone.clear_hint()
 self.__query_history[clone] += 1
 else:
 self.__query_history[clone] = 1
 self.__WriteHistory()
-results = [e._ToPb() for e in results]
 self.__cursor_lock.acquire()
 cursor = self.__next_cursor
 self.__next_cursor += 1
 self.__cursor_lock.release()
 self.__queries[cursor] = (results, len(results))
 query_result.mutable_cursor().set_cursor(cursor)
 query_result.set_more_results(len(results) > 0)
 def _Dynamic_Next(self, next_request, query_result):
 cursor = next_request.cursor().cursor()
 try:
 results, orig_count = self.__queries[cursor]
 except KeyError:
 raise apiproxy_errors.ApplicationError(datastore_pb.Error.BAD_REQUEST,
 'Cursor %d not found' % cursor)
 count = next_request.count()
-for r in results[:count]:
+results_pb = [r._ToPb() for r in results[:count]]
-query_result.add_result().CopyFrom(r)
+query_result.result_list().extend(results_pb)
 del results[:count]
 query_result.set_more_results(len(results) > 0)
 def _Dynamic_Count(self, query, integer64proto):
 query_result = datastore_pb.QueryResult()
 self._Dynamic_RunQuery(query, query_result)
 cursor = query_result.cursor().cursor()
 results, count = self.__queries[cursor]
 integer64proto.set_value(count)
 del self.__queries[cursor]
 def _Dynamic_BeginTransaction(self, request, transaction):
 self.__tx_handle_lock.acquire()
 handle = self.__next_tx_handle
 self.__next_tx_handle += 1
 self.__tx_handle_lock.release()
 kinds = []
 for app, kind in self.__entities:
 if app == app_str:
+app_kind = (app, kind)
+if app_kind in self.__schema_cache:
+kinds.append(self.__schema_cache[app_kind])
+continue
 kind_pb = entity_pb.EntityProto()
 kind_pb.mutable_key().set_app('')
 kind_pb.mutable_key().mutable_path().add_element().set_type(kind)
 kind_pb.mutable_entity_group()
-kinds.append(kind_pb)
 props = {}
-for entity in self.__entities[(app, kind)].values():
+for entity in self.__entities[app_kind].values():
-for prop in entity.native.property_list():
+for prop in entity.protobuf.property_list():
 if prop.name() not in props:
 props[prop.name()] = entity_pb.PropertyValue()
 props[prop.name()].MergeFrom(prop.value())
 for value_pb in props.values():
 for name, value_pb in props.items():
 prop_pb = kind_pb.add_property()
 prop_pb.set_name(name)
 prop_pb.mutable_value().CopyFrom(value_pb)
-schema.kind_list().extend(kinds)
+kinds.append(kind_pb)
+self.__schema_cache[app_kind] = kind_pb
+for kind_pb in kinds:
+schema.add_kind().CopyFrom(kind_pb)
 def _Dynamic_CreateIndex(self, index, id_response):
 if index.id() != 0:
 raise apiproxy_errors.ApplicationError(datastore_pb.Error.BAD_REQUEST,
 'New index id must be 0.')
 for stored_index in self.__indexes[app]:
 if index.definition() == stored_index.definition():
 return stored_index
 return None
+@classmethod
+def __GetSpecialPropertyValue(cls, entity, property):
+"""Returns an entity's value for a special property.
+Right now, the only special property is __key__, whose value is the
+entity's key.
+Args:
+entity: datastore.Entity
+Returns:
+property value. For __key__, a datastore_types.Key.
+Raises:
+AssertionError, if the given property is not special.
+"""
+assert property in datastore_types._SPECIAL_PROPERTIES
+if property == datastore_types._KEY_SPECIAL_PROPERTY:
+return entity.key()

changeset 686	df109be0567c
parent 297	35211afcd563
child 828	f5fd65cc3bf3