python类Key()的实例源码

def decode_key_property(obj, prop, value):
    if not value:
        return None

    return ndb.Key(urlsafe=value)

def test_but_missing(self):
        """
        .model is set to a key but the entity does not exist - pyamf must
        handle the encode and decode cases.
        """
        pyamf.register_class(models.Pet, 'pet')
        pyamf.register_class(models.SuperModel, 'supermodel')

        pet_key = ndb.Key('Pet', 'foobar')
        model_key = ndb.Key('SuperModel', 'barfoo')

        self.assertIsNone(model_key.get())

        pet = models.Pet(key=pet_key, model=model_key)

        pet.put()

        bytes = (
            b'\n\x0b\x07pet', (
                b'\t_key\x06Uagx0ZXN0YmVkLXRlc3RyDwsSA1BldCIGZm9vYmFyDA',
                b'\x0bmodel\x06gagx0ZXN0YmVkLXRlc3RyFgsSClN1cGVyTW9kZWwiBmJhcm'
                b'Zvbww',
                b'\tname\x01',
            ),
            b'\x01'
        )

        self.assertEncodes(pet, bytes)
        self.assertDecodes(bytes, pet)

def test_person(self):
        pyamf.register_class(models.Person, 'Person')

        person_key = ndb.Key('Contact', 'p')
        person = models.Person(
            key=person_key,
            first_name='Foo',
            last_name='Bar',
        )

        bytes = (
            b'\n\x0b\rPerson', (
                b'\t_key\x06Qagx0ZXN0YmVkLXRlc3RyDgsSB0NvbnRhY3QiAXAM',
                b'\x19phone_number\x01',
                b'\x0faddress\x01',
                b'\x15first_name\x06\x07Foo',
                b'\x13last_name\x06\x07Bar',
            ),
            b'\x01'
        )

        self.assertEncodes(person, bytes)

        def check_person(ret):
            self.assertIsInstance(ret, models.Person)
            self.assertEqual(ret, person)

        self.assertDecodes(bytes, check_person)

def test_company(self):
        pyamf.register_class(models.Company, 'Company')

        company_key = ndb.Key('Contact', 'c')

        company = models.Company(
            key=company_key,
            name='Acme Ltd',
        )

        bytes = (
            b'\n\x0b\x0fCompany', (
                b'\t_key\x06Qagx0ZXN0YmVkLXRlc3RyDgsSB0NvbnRhY3QiAWMM',
                b'\x19phone_number\x01',
                b'\x0faddress\x01',
                b'\tname\x06\x11Acme Ltd',
            ),
            b'\x01'
        )

        self.assertEncodes(company, bytes)

        def check_company(ret):
            self.assertIsInstance(ret, models.Company)
            self.assertEqual(ret, company)

        self.assertDecodes(bytes, check_company)

def test_structured(self):
        pyamf.register_class(models.SPContact, 'SPContact')
        pyamf.register_class(models.SPAddress, 'SPAddress')

        guido_key = ndb.Key('SPContact', 'guido')
        guido = models.SPContact(
            key=guido_key,
            name='Guido',
            addresses=[
                models.SPAddress(
                    type='home',
                    city='Amsterdam'
                ),
                models.SPAddress(
                    type='work',
                    street='Spear St',
                    city='SF'
                )
            ]
        )

        guido.put()

        bytes = (
            b'\n\x0b\x13SPContact\t_key\x06aagx0ZXN0YmVkLXRlc3RyFAsSCVNQQ29udG'
            b'FjdCIFZ3VpZG8M\x13addresses\t\x05\x01\n\x0b\x13SPAddress\x02\x01'
            b'\tcity\x06\x13Amsterdam\rstreet\x01\ttype\x06\thome\x01\n\x05'
            b'\x02\x01\n\x06\x05SF\x0e\x06\x11Spear St\x10\x06\twork\x01\tname'
            b'\x06\x0bGuido\x01'
        )

        self.assertEncodes(guido, bytes)

        def check_guido(ret):
            self.assertIsInstance(ret, models.SPContact)
            self.assertEqual(ret.key, guido_key)
            self.assertEqual(ret, guido)

            self.assertEqual(guido.addresses, ret.addresses)

        self.assertDecodes(bytes, check_guido)

def test_localstructured(self):
        pyamf.register_class(models.LSPContact, 'LSPContact')
        pyamf.register_class(models.SPAddress, 'SPAddress')

        guido_key = ndb.Key('LSPContact', 'guido')
        guido = models.LSPContact(
            key=guido_key,
            name='Guido',
            addresses=[
                models.SPAddress(
                    type='home',
                    city='Amsterdam'
                ),
                models.SPAddress(
                    type='work',
                    street='Spear St',
                    city='SF'
                )
            ]
        )

        guido.put()

        bytes = (
            b'\n\x0b\x15LSPContact\t_key\x06eagx0ZXN0YmVkLXRlc3RyFQsSCkxTUENv'
            b'bnRhY3QiBWd1aWRvDA\x13addresses\t\x05\x01\n\x0b\x13SPAddress\tc'
            b'ity\x06\x13Amsterdam\x02\x01\rstreet\x01\ttype\x06\thome\x01\n'
            b'\x05\n\x06\x05SF\x02\x01\x0e\x06\x11Spear St\x10\x06\twork\x01\t'
            b'name\x06\x0bGuido\x01'
        )

        self.assertEncodes(guido, bytes)

        def check_guido(ret):
            self.assertIsInstance(ret, models.LSPContact)
            self.assertEqual(ret.key, guido_key)
            self.assertEqual(ret, guido)

            self.assertEqual(guido.addresses, ret.addresses)

        self.assertDecodes(bytes, check_guido)

def __init__(self, *args, **kwargs):
        self.love = ndb.Key(Love, kwargs.get('love_id')).get()

def _post_delete_hook(cls, key, future):
        """
        Deletes an EmailReport from our search index when it is deleted from
        the datastore.

        Args:
            cls - The model class
            key - The instance ndb.Key
            future - The async future operation
        """
        future.check_success()
        search_provider.delete(str(key.id()))

def get_by_id(identifier):
        key = ndb.Key(PrivateNote, identifier)
        return key.get()

def get_by_id(identifier):
        key = ndb.Key(PrivateNote, identifier)
        return key.get()

def GetPriceByPredictionId(prediction_id):
  """Returns the current market price of contract one for a prediction_id."""
  prediction_key = ndb.Key(urlsafe=prediction_id)
  pred = prediction_key.get()
  price = math.pow(math.e, (pred.contract_one / pred.liquidity)) / (
      math.pow(math.e, (pred.contract_two / pred.liquidity)) + math.pow(
          math.e, (pred.contract_two / pred.liquidity)))
  return float(price)

def GetPrices(prediction_id):
    prices = Price.query(Price.prediction_id == ndb.Key(urlsafe=prediction_id)).order(-Price.date).fetch(30)
    print(str(prices))
    prices = [{'price': p.value, 'date': {'year': p.date.year, 'month': p.date.month, 'day': p.date.day}} for p in prices]
    return prices

def GetPredictionById(prediction_id):
  """Returns a prediction by prediction_id."""
  try:
    prediction_key = ndb.Key(urlsafe=prediction_id)
    prediction = prediction_key.get()
    portfolio = GetUserPortfolioByAuth(prediction_id)
  except:
    return render_template('404.html')
  return render_template(
      'prediction.html',
      prediction=prediction,
      price=GetPriceByPredictionId(prediction_id),
      pricelist=GetPrices(prediction_id),
      prediction_id=prediction_id,
      portfolio=portfolio)

def GetUserBalanceByAuth():
  """Returns current users balance."""
  user_key = ndb.Key('Profile', users.get_current_user().user_id())
  profile = user_key.get()
  return str(profile.balance)

# TODO(goldhaber): change to GetUserPortfolioByAuth By Prediction ID

def CreateTrade():
  """Creates a trade for the user."""
  user_id = users.get_current_user().user_id()
  user_key = ndb.Key('Profile', user_id)
  current_user = user_key.get()
  prediction_key = ndb.Key(urlsafe=request.form['prediction_id'])
  prediction = prediction_key.get()
  if request.form['is_likelihood'] == 'true':
    user_id = users.get_current_user().user_id()
    user_key = ndb.Key('Profile', user_id)
    current_user = user_key.get()
    trade = calculate_trade_from_likelihood(
        float(request.form['likelihood']), prediction, current_user)
    print trade
  else:
    trade = Trade(
        prediction_id=prediction_key,
        user_id=user_key,
        direction=request.form['direction'],
        contract=request.form['contract'],
        quantity=float(request.form['quantity']))
  err = CreateTradeAction(prediction, current_user, trade)
  #TODO replace with error
  if err != 'error':
      flash('You successfully predicted!')
  return redirect('/predictions/' + trade.prediction_id.urlsafe())

def GetTradesForPredictionId(prediction_id):
    user = users.get_current_user()
    trades = Trade.query(ndb.AND(Trade.prediction_id == ndb.Key(urlsafe=prediction_id),
                                 Trade.user_id == ndb.Key('Profile', user.user_id()))).fetch()
    return str(trades)

def inject_balance():
    user = users.get_current_user()
    if not user:
        return dict(balance=0)
    user_key = ndb.Key('Profile', user.user_id())
    profile = user_key.get()
    return dict(balance=profile.balance)

def GetEntityViaMemcache(entity_key):
  """Get entity from memcache if available, from datastore if not."""
  entity = memcache.get(entity_key)
  if entity is not None:
    return entity
  key = ndb.Key(urlsafe=entity_key)
  entity = key.get()
  if entity is not None:
    memcache.set(entity_key, entity)
  return entity

def GetPriceByPredictionId(prediction_id):
  """Get current price by Prediction Id"""
  prediction_key = ndb.Key(urlsafe=prediction_id)
  pred = prediction_key.get()
  price = math.pow(math.e, (pred.contract_one / pred.liquidity)) / (
      math.pow(math.e, (pred.contract_two / pred.liquidity)) + math.pow(
          math.e, (pred.contract_two / pred.liquidity)))
  return float(price)