def finish(self):
        # Refresh object
        self = self._meta.model.objects.get(pk=self.pk)

        # If this was called before, don't do anything
        if self.status == ImportStatus.FINISHED:
            return

        if self.get_meta().generate_error_csv:
            self.error_csv_filename = self._error_csv_filename()

            with cloudstorage.open(self.error_csv_filename, 'w') as f:
                # Concat all error csvs from shards into 1 file
                has_written = False

                shards = self.get_shard_model().objects.filter(task_id=self.pk, task_model_path=self.model_path)

                # The shards haven't necessarily finished writing their error files when this is called,
                # because that happens in a defer. So we redefer this until they're all done.
                if [shard for shard in shards if not shard.error_csv_written]:
                    self.defer(self.finish)
                    return

                for shard in shards:
                    if not shard.error_csv_filename:
                        continue

                    # If this is the first row, write the column headers
                    if not has_written:
                        data = json.loads(shard.source_data_json)[0]
                        cols = getattr(self, "detected_columns", data.keys()) + [ "errors" ]
                        csvwriter = csv.writer(f)
                        csvwriter.writerow(cols)
                        has_written = True

                    # Write the shard's error file into the master file
                    f.write(cloudstorage.open(shard.error_csv_filename).read())
                    cloudstorage.delete(shard.error_csv_filename)

            if has_written:
                # Create a blobstore key for the GCS file
                blob_key = create_gs_key('/gs%s' % self.error_csv_filename)
                self.error_csv = '%s/errors.csv' % blob_key
            else:
                cloudstorage.delete(self.error_csv_filename)

        self.status = ImportStatus.FINISHED
        self.save()
Called when all shards have finished processing
dataset/ETHPy150Open potatolondon/osmosis/osmosis/models.py/AbstractImportTask.finish