Skip to content

Instantly share code, notes, and snippets.

@brianrodri
Created October 19, 2020 17:46
Show Gist options
  • Save brianrodri/ef75c17dbf7b643de2530969d9819147 to your computer and use it in GitHub Desktop.
Save brianrodri/ef75c17dbf7b643de2530969d9819147 to your computer and use it in GitHub Desktop.
class BaseHumanMaintainedModel(BaseModel):
"""A model that tracks the last time it was updated by a human.
When the model is updated by a human, use `self.put_by_human()` to store it.
For example: after the title of an exploration is changed by its creator.
Otherwise, when the model is updated by some non-human or automated process,
use `self.put()`. For example: by a one-off job updating the schema version
of a property.
"""
# When this entity was last updated on behalf of a human.
last_updated_by_human = (
datastore_services.DateTimeProperty(indexed=True, required=True))
def put(self):
"""Unsupported operation on human-maintained models."""
self.last_updated_by_human = datetime.datetime.utcnow()
raise NotImplementedError('Use put_for_human or put_for_bot instead')
def put_async(self):
"""Unsupported operation on human-maintained models."""
self.last_updated_by_human = datetime.datetime.utcnow()
raise NotImplementedError(
'Use put_async_for_human or put_async_for_bot instead')
def put_for_human(self):
"""Stores the model instance on behalf of a human."""
self.last_updated_by_human = datetime.datetime.utcnow()
return super(self, BaseHumanMaintainedModel).put()
def put_async_for_human(self):
"""Stores the model instance asynchronously on behalf of a human."""
self.last_updated_by_human = datetime.datetime.utcnow()
return super(self, BaseHumanMaintainedModel).put_async()
def put_for_bot(self):
"""Stores the model instance on behalf of a non-human."""
return super(self, BaseHumanMaintainedModel).put()
def put_async_for_bot(self):
"""Stores the model instance asynchronously on behalf of a non-human."""
return super(self, BaseHumanMaintainedModel).put_async()
@classmethod
def put_multi_for_human(cls, instances):
"""Stores the given model instances on behalf of a human.
Args:
instances: list(HumanMaintainedModel). The instances to store.
"""
now = datetime.datetime.utcnow()
for instance in instances:
instance.last_updated_by_human = now
return cls.put_multi(instances)
@classmethod
def put_multi_async_for_human(cls):
"""Stores the given model instances asynchronously on behalf of a human.
Args:
instances: list(HumanMaintainedModel). The instances to store.
Returns:
list(future). A list of futures.
"""
now = datetime.datetime.utcnow()
for instance in instances:
instance.last_updated_by_human = now
return cls.put_multi_async(instances)
@classmethod
def put_multi_for_bot(cls, instances):
"""Stores the given model instances on behalf of a non-human.
Args:
instances: list(HumanMaintainedModel). The instances to store.
"""
return cls.put_multi(instances)
@classmethod
def put_multi_async_for_bot(cls):
"""Stores the given model instances asynchronously on behalf of a
non-human.
Args:
instances: list(HumanMaintainedModel). The instances to store.
Returns:
list(future). A list of futures.
"""
return cls.put_multi_async(instances)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment