def process_chunk(self):
        def _assert_change_has_id(change):
            if 'id' not in change:
                _assert(False, "expected 'id' in change, but wasn't found! change is: {}".format(
                    simplejson.dumps(change)
                ))
                return False
            return True

        changes_to_process = filter(_assert_change_has_id, self.change_queue)
        if self.preload_docs:
            self.get_couch_db().bulk_load([change['id'] for change in changes_to_process],
                                     purge_existing=True)
        for change in changes_to_process:
            if self.preload_docs:
                doc = self.get_couch_db().open_doc(change['id'], check_main=False)
                change.set_document(doc)

            # a valid change is either a non-preload situation or a valid doc + a filter match
            valid_change = (not self.preload_docs or change.document) and self.python_filter(change)
            valid_deletion = self.process_deletions and change.get('deleted', None)
            if valid_change or valid_deletion:
                try:
                    self.process_change(change)
                except Exception:
                    logging.exception('something went wrong processing change %s (%s)' %
                                      (change.get('seq', None), change['id']))

        # reset the queue after we've processed this chunk
        self.change_queue = []
        self.last_processed_time = datetime.utcnow()
dataset/ETHPy150Open dimagi/commcare-hq/corehq/ex-submodules/pillowtop/listener.py/PythonPillow.process_chunk/original