return_progress as arg for run_query

cdci_data_analysis/analysis/instrument.py

@@ -390,6 +390,7 @@ def run_query(self, product_type,
                   dry_run=False,
                   api=False,
                   decoded_token=None,
+                  return_progress=False,
                   **kwargs):
 
         if logger is None:
@@ -433,6 +434,7 @@ def run_query(self, product_type,
                                                     config=config,
                                                     logger=logger,
                                                     sentry_dsn=sentry_dsn,
+                                                    return_progress=return_progress,
                                                     api=api)
                     if query_out.status_dictionary['status'] == 0:
                         if 'comment' in query_out.status_dictionary.keys():

cdci_data_analysis/analysis/queries.py

@@ -380,6 +380,12 @@
     def get_dummy_products(self,instrument, config=None,**kwargs):
         raise RuntimeError(f'{self}: get_dummy_products needs to be implemented in derived class')
 
+    def get_progress(self):
+        raise RuntimeError(f'{self}: get_progress needs to be implemented in derived class')
+
+    def get_dummy_progress(self, instrument, config=None,**kwargs):
+        raise RuntimeError(f'{self}: get_dummy_progress needs to be implemented in derived class')
+
     def get_data_server_query(self,instrument,config=None,**kwargs):
         traceback.print_stack()
         raise RuntimeError(f'{self}: get_data_server_query needs to be implemented in derived class')
@@ -526,6 +532,35 @@
 
         return query_out
 
+    def get_query_progress_details(self, instrument, job, query_type='Real', config=None, logger=None, scratch_dir=None, api=False):
+        if logger is None:
+            logger = self.get_logger()
+
+        query_out = QueryOutput()
+        # status=0
+
+        messages = {
+            'message': '',
+            'debug_message': '',
+            'comment': '',
+            'warning': ''
+        }
+        try:
+            if query_type != 'Dummy':
+                q = self.get_data_server_query(instrument, config)
+                q.get_progress()
+            else:
+                status = 0
+                self.query_prod_list = self.get_dummy_progress(instrument,
+                                                                       config=config,
+                                                                       out_dir=scratch_dir,
+                                                                       api=api)
+
+                job.set_done()
+        except Exception as e:
+            sentry_sdk.capture_exception(e)
+            raise InternalError(f"unexpected error while getting query progress details with {instrument}, {e}")
+
     def get_query_products(self,instrument,job,run_asynch,query_type='Real',logger=None,config=None,scratch_dir=None,sentry_dsn=None,api=False):
         if logger is None:
             logger = self.get_logger()
@@ -545,7 +580,9 @@
             if query_type != 'Dummy':
                 q = self.get_data_server_query(instrument,config)
 
-                res, data_server_query_out = q.run_query(call_back_url=job.get_call_back_url(), run_asynch=run_asynch, logger=logger)
+                res, data_server_query_out = q.run_query(call_back_url=job.get_call_back_url(),
+                                                         run_asynch=run_asynch,
+                                                         logger=logger)
 
                 for field in ['message', 'debug_message', 'comment', 'warning']:
                     if field in data_server_query_out.status_dictionary.keys():
@@ -570,7 +607,10 @@
 
             else:
                 status=0
-                self.query_prod_list = self.get_dummy_products(instrument,config=config,out_dir=scratch_dir,api=api)
+                self.query_prod_list = self.get_dummy_products(instrument,
+                                                               config=config,
+                                                               out_dir=scratch_dir,
+                                                               api=api)
 
                 #self.query_prod_list = QueryProductList(prod_list=prod_list)
 
@@ -677,7 +717,8 @@
                   config=None, 
                   logger=None,
                   sentry_dsn=None,
-                  api=False):
+                  api=False,
+                  return_progress=False):
 
         # print ('--> running query for ',instrument.name,'with config',config)
         if logger is None:
@@ -700,18 +741,36 @@
 
 
         if query_out.status_dictionary['status'] == 0:
-            query_out = self.get_query_products(instrument,job,run_asynch, query_type=query_type, logger=logger, config=config, scratch_dir=scratch_dir, sentry_dsn=sentry_dsn, api=api)
+            query_out = self.get_query_products(instrument,
+                                                job,
+                                                run_asynch,
+                                                query_type=query_type,
+                                                logger=logger,
+                                                config=config,
+                                                scratch_dir=scratch_dir,
+                                                sentry_dsn=sentry_dsn,
+                                                api=api)
             self._t_query_steps['after_get_query_products'] = _time.time()
 
         if query_out.status_dictionary['status'] == 0:
-            if job.status!='done':
+            if job.status != 'done':
 
                 query_out.prod_dictionary = {}
                 # TODO: add check if is asynch
                 # TODO: the asynch status will be in the qery_out class
                 # TODO: if asynch and running return proper query_out
                 # TODO: if asynch and done proceed
 
+                if return_progress:
+                    self.get_query_progress_details(instrument,
+                                                    job,
+                                                    query_type=query_type,
+                                                    logger=logger,
+                                                    config=config,
+                                                    scratch_dir=scratch_dir,
+                                                    api=api
+                                                    )
+
             else:
                 if query_out.status_dictionary['status'] == 0:
                     #print('-->',query_out.status_dictionary)

cdci_data_analysis/flask_app/dispatcher_query.py

@@ -161,6 +161,7 @@ def __init__(self, app,
 
 
             self.client_name = self.par_dic.pop('client-name', 'unknown')
+            self.return_progress = self.par_dic.pop('return-progress', False) == 'True'
             if os.environ.get("DISPATCHER_ASYNC_ENABLED", "no") == "yes":  # TODO: move to config!
                 self.async_dispatcher = self.par_dic.pop(
                     'async_dispatcher', 'True') == 'True'  # why string true?? else false anyway
@@ -1909,6 +1910,7 @@ def run_query(self, off_line=False, disp_conf=None):
                                                           verbose=verbose,
                                                           dry_run=dry_run,
                                                           api=api,
+                                                          return_progress=self.return_progress,
                                                           decoded_token=self.decoded_token)
                     self.log_query_progression("after instrument.run_query")                                                          
                 except RequestNotAuthorized as e:

cdci_data_analysis/plugins/dummy_plugin/data_server_dispatcher.py

@@ -104,26 +104,21 @@
 
         status = self.decide_status() 
         if status == "submitted":
-            # set done to submitted?
             query_out.set_done(message="job submitted mock",
-                            debug_message="no message really",
-                            job_status='submitted',
-                            comment="mock comment",
-                            warning="mock warning")
+                               debug_message="no message really",
+                               job_status='submitted',
+                               comment="mock comment",
+                               warning="mock warning")
         elif status == "done":
-            # set done to submitted?
             query_out.set_done(message="job done mock",
-                            debug_message="no message really",
-                            job_status='done',
-                            comment="mock comment",
-                            warning="mock warning")
+                               debug_message="no message really",
+                               job_status='done',
+                               comment="mock comment",
+                               warning="mock warning")
         elif status == "failed":
-            # set done to submitted?
             query_out.set_failed(message="job failed mock",
-                            debug_message="no message really",
-                            job_status='failed',
-                            comment="mock comment",
-                            warning="mock warning")
+                                 debug_message="no message really",
+                                 job_status='failed')
         else:
             NotImplementedError
 
@@ -164,6 +159,34 @@
         return None, query_out
 
 
+class ReturnProgressProductQuery(ProductQuery):
+    def __init__(self, name, parameters_list=None):
+        if parameters_list is None:
+            parameters_list = []
+        super().__init__(name, return_progress=True, parameters_list=parameters_list)
+
+
+    def run_query(self,
+                  instrument,
+                  scratch_dir,
+                  job,
+                  run_asynch,
+                  query_type='Real',
+                  config=None,
+                  logger=None,
+                  sentry_dsn=None,
+                  api=False,
+                  return_progress=False):
+        query_out = self.process_query_product(instrument, job, logger=logger, config=config, scratch_dir=scratch_dir,
+                                               sentry_dsn=sentry_dsn, api=api)
+        if query_out.status_dictionary['status'] == 0:
+            job.set_done()
+        else:
+            job.set_failed()
+
+        return query_out
+
+
 class EmptyProductQuery(ProductQuery):
 
     def __init__(self, name='unset-name', config=None, instrument=None):