+ if o.cares_about(delta):
+ asyncio.ensure_future(o(delta, old_obj, new_obj, self))
+
+ async def _wait(self, entity_type, entity_id, action, predicate=None):
+ """
+ Block the calling routine until a given action has happened to the
+ given entity
+
+ :param entity_type: The entity's type.
+ :param entity_id: The entity's id.
+ :param action: the type of action (e.g., 'add' or 'change')
+ :param predicate: optional callable that must take as an
+ argument a delta, and must return a boolean, indicating
+ whether the delta contains the specific action we're looking
+ for. For example, you might check to see whether a 'change'
+ has a 'completed' status. See the _Observer class for details.
+
+ """
+ q = asyncio.Queue(loop=self.loop)
+
+ async def callback(delta, old, new, model):
+ await q.put(delta.get_id())
+
+ self.add_observer(callback, entity_type, action, entity_id, predicate)
+ entity_id = await q.get()
+ return self.state._live_entity_map(entity_type)[entity_id]
+
+ async def _wait_for_new(self, entity_type, entity_id=None, predicate=None):
+ """Wait for a new object to appear in the Model and return it.
+
+ Waits for an object of type ``entity_type`` with id ``entity_id``.
+ If ``entity_id`` is ``None``, it will wait for the first new entity
+ of the correct type.
+
+ This coroutine blocks until the new object appears in the model.
+
+ """
+ # if the entity is already in the model, just return it
+ if entity_id in self.state._live_entity_map(entity_type):
+ return self.state._live_entity_map(entity_type)[entity_id]
+ # if we know the entity_id, we can trigger on any action that puts
+ # the enitty into the model; otherwise, we have to watch for the
+ # next "add" action on that entity_type
+ action = 'add' if entity_id is None else None
+ return await self._wait(entity_type, entity_id, action, predicate)
+
+ async def wait_for_action(self, action_id):
+ """Given an action, wait for it to complete."""
+
+ if action_id.startswith("action-"):
+ # if we've been passed action.tag, transform it into the
+ # id that the api deltas will use.
+ action_id = action_id[7:]
+
+ def predicate(delta):
+ return delta.data['status'] in ('completed', 'failed')
+
+ return await self._wait('action', action_id, 'change', predicate)