@@ -0,0 +1,1040 @@
+from pipelines.models import Operation%0Afrom polyaxon.celery_api import CeleryTask%0A%0A%0Aclass OperationTask(CeleryTask):%0A    %22%22%22Base operation celery task with basic logging.%22%22%22%0A    _operation = None%0A%0A    def run(self, *args, **kwargs):%0A        self._operation = Operation.objects.get(id=kwargs%5B'query_id'%5D)%0A        super(OperationTask, self).run(*args, **kwargs)%0A%0A    def on_failure(self, exc, task_id, args, kwargs, einfo):%0A        %22%22%22Update query status and send email notification to a user%22%22%22%0A        super(OperationTask, self).on_failure(exc, task_id, args, kwargs, einfo)%0A        self._operation.on_failure()%0A%0A    def on_retry(self, exc, task_id, args, kwargs, einfo):%0A        super(OperationTask, self).on_retry(exc, task_id, args, kwargs, einfo)%0A        self._operation.on_retry()%0A%0A    def on_success(self, retval, task_id, args, kwargs):%0A        %22%22%22Send email notification and a file, if requested to do so by a user%22%22%22%0A        super(OperationTask, self).on_success(retval, task_id, args, kwargs)%0A        self._operation.on_success()%0A

Add base celery operation task