def process(self):
        meta = self.task.get_meta()
        task_model = get_model(*self.task.model_path.split("."))

        this = ImportShard.objects.get(pk=self.pk)  # Reload, self is pickled
        source_data = json.loads(this.source_data_json)

        # If there are no rows to process
        mark_shard_complete = this.last_row_processed == this.total_rows - 1 or this.total_rows == 0

        for i in xrange(this.last_row_processed, this.total_rows):  # Always continue from the last processed row
            data = source_data[i]

            forms = [self.task.instantiate_form(form, data) for form in meta.forms]

            if all([form.is_valid() for form in forms]):
                # All forms are valid, let's process this shizzle

                cleaned_data = {}
                for form in forms:
                    cleaned_data.update(form.cleaned_data)

                try:
                    self.task.import_row(forms, cleaned_data)
                except __HOLE__, e:
                    # We allow subclasses to raise a validation error on import_row
                    errors = []
                    if hasattr(e, 'message_dict'):
                        for name, errs in e.message_dict.items():
                            for err in errs:
                                errors.append("{0}: {1}".format(name, err))
                    else:
                        # Pre 1.6, ValidationError does not necessarily have a message_dict
                        for err in e.messages:
                            errors.append(err)

                    self.handle_error(this.start_line_number + i, cleaned_data, errors)
            else:
                # We've encountered an error, call the error handler
                errors = []
                for form in forms:
                    for name, errs in form.errors.items():
                        for err in errs:
                            errors.append("{0}: {1}".format(name, err))

                self.handle_error(this.start_line_number + i, data, errors)

            # Now update the last processed row, transactionally
            @transactional
            def update_shard(_this):
                _this = ImportShard.objects.get(pk=_this.pk)
                _this.last_row_processed += 1
                _this.save()
                return _this

            this = update_shard(this)
            # If this was the last iteration then mark as complete
            mark_shard_complete = i == this.total_rows - 1

        if mark_shard_complete:
            @transactional
            def update_task(_this):
                if _this.complete:
                    return

                task = task_model.objects.get(pk=_this.task_id)
                task.shards_processed += 1
                task.save()

                _this.complete = True
                _this.save()

            update_task(this)
            deferred.defer(this._finalize_errors, _queue=self.task.get_meta().queue)
dataset/ETHPy150Open potatolondon/osmosis/osmosis/models.py/ImportShard.process