Source code for oauth_dropins.webutil.models
"""App Engine datastore model base classes and utilites.
"""
__author__ = ['Ryan Barrett <webutil@ryanb.org>']
import functools
from google.appengine.ext import db
from google.appengine.ext import ndb
[docs]class StringIdModel(ndb.Model):
"""An ndb model class that requires a string id."""
[docs] def put(self, *args, **kwargs):
"""Raises AssertionError if string id is not provided."""
assert self.key and self.key.string_id(), 'string id required but not provided'
return super(StringIdModel, self).put(*args, **kwargs)
[docs]class KeyNameModel(db.Model):
"""A db model class that requires a key name."""
[docs] def __init__(self, *args, **kwargs):
"""Raises AssertionError if key name is not provided."""
super(KeyNameModel, self).__init__(*args, **kwargs)
try:
assert self.key().name()
except db.NotSavedError:
assert False, 'key name required but not provided'
[docs] def __str__(self):
return self.key().to_path()
[docs]class SingleEGModel(db.Model):
"""A model class that stores all entities in a single entity group.
All entities use the same parent key (below), and :meth:`all()` automatically
adds it as an ancestor. That allows, among other things, fetching all entities
of this kind with strong consistency.
"""
[docs] def enforce_parent(fn):
"""Sets the parent keyword arg. If it's already set, checks that it's correct."""
@functools.wraps(fn)
def wrapper(self_or_cls, *args, **kwargs):
if '_from_entity' not in kwargs:
parent = self_or_cls.shared_parent_key()
if 'parent' in kwargs:
assert kwargs['parent'] == parent, "Can't override parent in SingleEGModel"
kwargs['parent'] = parent
return fn(self_or_cls, *args, **kwargs)
return wrapper
[docs] @classmethod
def shared_parent_key(cls):
"""Returns the shared parent key for this class.
It's not actually an entity, just a placeholder key.
"""
return db.Key.from_path('Parent', cls.kind())
[docs] @enforce_parent
def __init__(self, *args, **kwargs):
super(SingleEGModel, self).__init__(*args, **kwargs)
[docs] @classmethod
@enforce_parent
def get_by_id(cls, id, **kwargs):
return super(SingleEGModel, cls).get_by_id(id, **kwargs)
[docs] @classmethod
@enforce_parent
def get_by_key_name(cls, key_name, **kwargs):
return super(SingleEGModel, cls).get_by_key_name(key_name, **kwargs)
[docs] @classmethod
@enforce_parent
def get_or_insert(cls, key_name, **kwargs):
return super(SingleEGModel, cls).get_or_insert(key_name, **kwargs)
[docs] @classmethod
def all(cls):
return db.Query(cls).ancestor(cls.shared_parent_key())