我们从Python开源项目中，提取了以下48个代码示例，用于说明如何使用google.appengine.ext.ndb.Model()。
def _is_ndb(self): """Determine whether the model of the instance is an NDB model. Returns: Boolean indicating whether or not the model is an NDB or DB model. """ # issubclass will fail if one of the arguments is not a class, only # need worry about new-style classes since ndb and db models are # new-style if isinstance(self._model, type): if ndb is not None and issubclass(self._model, ndb.Model): return True elif issubclass(self._model, db.Model): return False raise TypeError('Model class not an NDB or DB model: %s.' % (self._model,))
def encode_ndb_key(key, encoder=None): """ When encountering an L{ndb.Key} instance, find the entity in the datastore and encode that. """ klass = ndb.Model._kind_map.get(key.kind()) referenced_object = _get_by_class_key( encoder, klass, key, ) if not referenced_object: encoder.writeElement(None) return encoder.writeObject(referenced_object)
def make_directed_query(self, kind_class, keys_only=False): """Construct a query for this key range, including the scan direction. Args: kind_class: A kind implementation class (a subclass of either db.Model or ndb.Model). keys_only: bool, default False, use keys_only on Query? Returns: A db.Query or ndb.Query instance (corresponding to kind_class). Raises: KeyRangeError: if self.direction is not in (KeyRange.ASC, KeyRange.DESC). """ if ndb is not None: if issubclass(kind_class, ndb.Model): return self.make_directed_ndb_query(kind_class, keys_only=keys_only) assert self._app is None, '_app is not supported for db.Query' direction = self.__get_direction("", "-") query = db.Query(kind_class, namespace=self.namespace, keys_only=keys_only) query.order("%s__key__" % direction) query = self.filter_query(query) return query
def make_ascending_query(self, kind_class, keys_only=False, filters=None): """Construct a query for this key range without setting the scan direction. Args: kind_class: A kind implementation class (a subclass of either db.Model or ndb.Model). keys_only: bool, default False, query only for keys. filters: optional list of filters to apply to the query. Each filter is a tuple: (<property_name_as_str>, <query_operation_as_str>, <value>). User filters are applied first. Returns: A db.Query or ndb.Query instance (corresponding to kind_class). """ if ndb is not None: if issubclass(kind_class, ndb.Model): return self.make_ascending_ndb_query( kind_class, keys_only=keys_only, filters=filters) assert self._app is None, '_app is not supported for db.Query' query = db.Query(kind_class, namespace=self.namespace, keys_only=keys_only) query.order("__key__") query = self.filter_query(query, filters=filters) return query
def make_ascending_ndb_query(self, kind_class, keys_only=False, filters=None): """Construct an NDB query for this key range, without the scan direction. Args: kind_class: An ndb.Model subclass. keys_only: bool, default False, query only for keys. Returns: An ndb.Query instance. """ assert issubclass(kind_class, ndb.Model) if keys_only: default_options = ndb.QueryOptions(keys_only=True) else: default_options = None query = kind_class.query(app=self._app, namespace=self.namespace, default_options=default_options) query = self.filter_ndb_query(query, filters=filters) query = query.order(kind_class._key) return query
def __init__(self, model, key_name, property_name, cache=None, user=None): """Constructor for Storage. Args: model: db.Model or ndb.Model, model class key_name: string, key name for the entity that has the credentials property_name: string, name of the property that is a CredentialsProperty or CredentialsNDBProperty. cache: memcache, a write-through cache to put in front of the datastore. If the model you are using is an NDB model, using a cache will be redundant since the model uses an instance cache and memcache for you. user: users.User object, optional. Can be used to grab user ID as a key_name if no key name is specified. """ if key_name is None: if user is None: raise ValueError('StorageByKeyName called with no key name or user.') key_name = user.user_id() self._model = model self._key_name = key_name self._property_name = property_name self._cache = cache
def __init__(self, model, key_name, property_name, cache=None, user=None): """Constructor for Storage. Args: model: db.Model or ndb.Model, model class key_name: string, key name for the entity that has the credentials property_name: string, name of the property that is a CredentialsProperty or CredentialsNDBProperty. cache: memcache, a write-through cache to put in front of the datastore. If the model you are using is an NDB model, using a cache will be redundant since the model uses an instance cache and memcache for you. user: users.User object, optional. Can be used to grab user ID as a key_name if no key name is specified. """ if key_name is None: if user is None: raise ValueError('StorageByKeyName called with no ' 'key name or user.') key_name = user.user_id() self._model = model self._key_name = key_name self._property_name = property_name self._cache = cache
def query_purchase_with_customer_key(): # [START purchase_with_customer_key_models] class Customer(ndb.Model): name = ndb.StringProperty() class Purchase(ndb.Model): customer = ndb.KeyProperty(kind=Customer) price = ndb.IntegerProperty() # [END purchase_with_customer_key_models] def query_purchases_for_customer_via_key(customer_entity): purchases = Purchase.query( Purchase.customer == customer_entity.key).fetch() return purchases return Customer, Purchase, query_purchases_for_customer_via_key
def query_purchase_with_ancestor_key(): # [START purchase_with_ancestor_key_models] class Customer(ndb.Model): name = ndb.StringProperty() class Purchase(ndb.Model): price = ndb.IntegerProperty() # [END purchase_with_ancestor_key_models] def create_purchase_for_customer_with_ancestor(customer_entity): purchase = Purchase(parent=customer_entity.key) return purchase def query_for_purchases_of_customer_with_ancestor(customer_entity): purchases = Purchase.query(ancestor=customer_entity.key).fetch() return purchases return (Customer, Purchase, create_purchase_for_customer_with_ancestor, query_for_purchases_of_customer_with_ancestor)
def testEventuallyConsistentGlobalQueryResult(self): class TestModel(ndb.Model): pass user_key = ndb.Key('User', 'ryan') # Put two entities ndb.put_multi([ TestModel(parent=user_key), TestModel(parent=user_key) ]) # Global query doesn't see the data. self.assertEqual(0, TestModel.query().count(3)) # Ancestor query does see the data. self.assertEqual(2, TestModel.query(ancestor=user_key).count(3)) # [END HRD_example_1] # [START HRD_example_2]
def testDeterministicOutcome(self): # 50% chance to apply. self.policy.SetProbability(.5) # Use the pseudo random sequence derived from seed=2. self.policy.SetSeed(2) class TestModel(ndb.Model): pass TestModel().put() self.assertEqual(0, TestModel.query().count(3)) self.assertEqual(0, TestModel.query().count(3)) # Will always be applied before the third query. self.assertEqual(1, TestModel.query().count(3)) # [END HRD_example_2] # [START main]
def _from_base_type(self, ent): """Convert a Model instance (entity) to a Message value.""" if ent._projection: # Projection query result. Reconstitute the message from the fields. return _projected_entity_to_message(ent, self._message_type) blob = ent.blob_ if blob is not None: protocol = self._protocol_impl else: # Perhaps it was written using a different protocol. protocol = None for name in _protocols_registry.names: key = '__%s__' % name if key in ent._values: blob = ent._values[key] if isinstance(blob, model._BaseValue): blob = blob.b_val protocol = _protocols_registry.lookup_by_name(name) break if blob is None or protocol is None: return None # This will reveal the underlying dummy model. msg = protocol.decode_message(self._message_type, blob) return msg
def _message_to_entity(msg, modelclass): """Recursive helper for _to_base_type() to convert a message to an entity. Args: msg: A Message instance. modelclass: A Model subclass. Returns: An instance of modelclass. """ ent = modelclass() for prop_name, prop in modelclass._properties.iteritems(): if prop._code_name == 'blob_': # TODO: Devise a cleaner test. continue # That's taken care of later. value = getattr(msg, prop_name) if value is not None and isinstance(prop, model.StructuredProperty): if prop._repeated: value = [_message_to_entity(v, prop._modelclass) for v in value] else: value = _message_to_entity(value, prop._modelclass) setattr(ent, prop_name, value) return ent
def _validate_filters_ndb(cls, filters, model_class): """Validate ndb.Model filters.""" if not filters: return properties = model_class._properties for f in filters: prop, _, val = f if prop not in properties: raise errors.BadReaderParamsError( "Property %s is not defined for entity type %s", prop, model_class._get_kind()) try: properties[prop]._do_validate(val) except db.BadValueError, e: raise errors.BadReaderParamsError(e)
def validate(cls, job_config): """Inherit docs.""" super(ModelDatastoreInputReader, cls).validate(job_config) params = job_config.input_reader_params entity_kind = params[cls.ENTITY_KIND_PARAM] try: model_class = util.for_name(entity_kind) except ImportError, e: raise errors.BadReaderParamsError("Bad entity kind: %s" % e) if cls.FILTERS_PARAM in params: filters = params[cls.FILTERS_PARAM] if issubclass(model_class, db.Model): cls._validate_filters(filters, model_class) else: cls._validate_filters_ndb(filters, model_class) property_range.PropertyRange(filters, entity_kind)
def get_by_urlsafe(self, urlsafe): """Returns an ndb.Model entity that the urlsafe key points to. Checks that the type of entity returned is of the correct kind. Raises an error if the key String is malformed or the entity is of the incorrect kind. :param urlsafe: A urlsafe key string """ try: key = ndb.Key(urlsafe=urlsafe) except TypeError: raise endpoints.BadRequestException('Invalid Key') except Exception, e: if e.__class__.__name__ == 'ProtocolBufferDecodeError': raise endpoints.BadRequestException('Invalid Key') else: raise model = key.get() if not model: return None self._isinstance(model) return model
def properties_to_json(self, dict_): """Converts ndb properties to their string representations for json. As a first step in converting ndb models to json, to_dict() is called. This is sufficient if all the ndb properties are strings. If not, subclasses should override this method to convert each field to its string representation. This base class implementation is a pass-through. Args: dict_: A deep copy of the dict as returned from ndb.Model.to_dict(). For simple transformations fo the fields, this object can be modified and returned. Returns: A json representation of the model. """ return dict_
def __init__(self, model, key_name, property_name, cache=None, user=None): """Constructor for Storage. Args: model: db.Model or ndb.Model, model class key_name: string, key name for the entity that has the credentials property_name: string, name of the property that is a CredentialsProperty or CredentialsNDBProperty. cache: memcache, a write-through cache to put in front of the datastore. If the model you are using is an NDB model, using a cache will be redundant since the model uses an instance cache and memcache for you. user: users.User object, optional. Can be used to grab user ID as a key_name if no key name is specified. """ super(StorageByKeyName, self).__init__() if key_name is None: if user is None: raise ValueError('StorageByKeyName called with no ' 'key name or user.') key_name = user.user_id() self._model = model self._key_name = key_name self._property_name = property_name self._cache = cache
def _is_ndb(self): """Determine whether the model of the instance is an NDB model. Returns: Boolean indicating whether or not the model is an NDB or DB model. """ # issubclass will fail if one of the arguments is not a class, only need # worry about new-style classes since ndb and db models are new-style if isinstance(self._model, type): if ndb is not None and issubclass(self._model, ndb.Model): return True elif issubclass(self._model, db.Model): return False raise TypeError('Model class not an NDB or DB model: %s.' % (self._model,))
def post_ndb_process(payload, context): """ """ stubs = context.get(NDB_STUB_NAME, None) if not stubs: return payload stubs.transform() return payload # small optimisation to compile the ndb.Model base class
def model_form(model, base_class=Form, only=None, exclude=None, field_args=None, converter=None): """ Creates and returns a dynamic ``wtforms.Form`` class for a given ``ndb.Model`` class. The form class can be used as it is or serve as a base for extended form classes, which can then mix non-model related fields, subforms with other model forms, among other possibilities. :param model: The ``ndb.Model`` class to generate a form for. :param base_class: Base form class to extend from. Must be a ``wtforms.Form`` subclass. :param only: An optional iterable with the property names that should be included in the form. Only these properties will have fields. :param exclude: An optional iterable with the property names that should be excluded from the form. All other properties will have fields. :param field_args: An optional dictionary of field names mapping to keyword arguments used to construct each field object. :param converter: A converter to generate the fields based on the model properties. If not set, ``ModelConverter`` is used. """ # Extract the fields from the model. field_dict = model_fields(model, only, exclude, field_args, converter) # Return a dynamically created form class, extending from base_class and # including the created fields as properties. return type(model._get_kind() + 'Form', (base_class,), field_dict)
def make_directed_ndb_query(self, kind_class, keys_only=False): """Construct an NDB query for this key range, including the scan direction. Args: kind_class: An ndb.Model subclass. keys_only: bool, default False, use keys_only on Query? Returns: An ndb.Query instance. Raises: KeyRangeError: if self.direction is not in (KeyRange.ASC, KeyRange.DESC). """ assert issubclass(kind_class, ndb.Model) if keys_only: default_options = ndb.QueryOptions(keys_only=True) else: default_options = None query = kind_class.query(app=self._app, namespace=self.namespace, default_options=default_options) query = self.filter_ndb_query(query) if self.__get_direction(True, False): query = query.order(kind_class._key) else: query = query.order(-kind_class._key) return query
