Merge pull request #232 from CBIIT/CRDCDH-2370-001-pgu

Crdcdh 2370

Author: n2iw

configs/messages_configuration.yml

@@ -130,6 +130,11 @@ Messages:
    severity: "Error"
    message: '"properties" is not defined in the model.'
    description: "Data model missing properties definitions."
+  M027:
+   title: "CDE not available"
+   severity: "Error"
+   message: '{} CDE for property "{}" is not available. Please try again later and contact the helpdesk if this error persists.'
+   description: "CDE version doesn't exist in DB, and on-demand pull failed."
 
   # data file validation message
   F001:

src/metadata_validator.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-import pandas as pd
 import json
 from datetime import datetime
 from bento.common.sqs import VisibilityExtender
@@ -18,10 +17,11 @@
 from common.model_reader import valid_prop_types
 from service.ecs_agent import set_scale_in_protection
 from x_submission_validator import CrossSubmissionValidator
-from pv_puller import get_pv_by_code_version
+from pv_puller import get_pv_by_datacommon_version_cde
 
 VISIBILITY_TIMEOUT = 20
 BATCH_SIZE = 1000
+CDE_NOT_FOUND = "CDE not available"
 
 def metadataValidate(configs, job_queue, mongo_dao):
     log = get_logger('Metadata Validation Service')
@@ -93,12 +93,6 @@ def metadataValidate(configs, job_queue, mongo_dao):
         except KeyboardInterrupt:
             log.info('Good bye!')
             return
-
-
-""" Requirement for the ticket crdcdh-343
-For metadata: validate data folder contains TSV or TXT files
-Compose a list of files to be updated and their sizes (metadata or files)
-"""
 class MetaDataValidator:
 
     def __init__(self, mongo_dao, model_store, config):
@@ -112,6 +106,8 @@ def __init__(self, mongo_dao, model_store, config):
         self.submission = None
         self.isError = None
         self.isWarning = None
+        self.searched_sts = False
+        self.not_found_cde = False
 
 
     def validate(self, submission_id, scope):
@@ -487,7 +483,9 @@ def validate_prop_value(self, prop_name, value, prop_def, msg_prefix):
             val = None
             minimum = prop_def.get(MIN)
             maximum = prop_def.get(MAX)
-            permissive_vals = self.get_permissive_value(prop_def)
+            permissive_vals, msg = self.get_permissive_value(prop_def)
+            if msg and msg == CDE_NOT_FOUND:
+                errors.append(create_error("M027", [msg_prefix, prop_name], prop_name, value))
             if type == "string":
                 val = str(value)
                 result, error = check_permissive(val, permissive_vals, msg_prefix, prop_name, self.mongo_dao)
@@ -520,12 +518,6 @@ def validate_prop_value(self, prop_name, value, prop_def, msg_prefix):
                 if len(errs) > 0:
                     errors.extend(errs)
 
-            # elif type == "datetime":
-            #     try:
-            #         val = datetime.strptime(value, DATETIME_FORMAT)
-            #     except ValueError as e:
-            #         errors.append(create_error("Invalid datetime value", f'{msg_prefix} Property "{prop_name}": "{value}" is not a valid datetime type.'))
-
             elif type == "date" or type == "datetime":
                 val = None
                 for date_format in DATE_FORMATS:
@@ -562,6 +554,7 @@ def validate_prop_value(self, prop_name, value, prop_def, msg_prefix):
     """
     def get_permissive_value(self, prop_def):
         permissive_vals = prop_def.get("permissible_values") 
+        msg = None
         if prop_def.get(CDE_TERM) and len(prop_def.get(CDE_TERM)) > 0:
             # retrieve permissible values from DB or cde site
             cde_code = None
@@ -580,19 +573,28 @@ def get_permissive_value(self, prop_def):
                     else:
                         permissive_vals = None
             else:
-                # call pv_puller to get permissible values from caDSR
-                cde, msg = get_pv_by_code_version(self.config, self.log, self.datacommon, prop_def["name"], cde_code, cde_version)
-                if cde:
-                    if cde.get(CDE_PERMISSIVE_VALUES) is not None:
-                        if len(cde[CDE_PERMISSIVE_VALUES]) > 0:                        
-                            permissive_vals = cde[CDE_PERMISSIVE_VALUES]
-                        else:
-                            permissive_vals =  None #escape validation
-                    self.mongo_dao.insert_cde([cde])  
+                if not self.searched_sts:
+                    cde = get_pv_by_datacommon_version_cde(self.config[TIER_CONFIG], self.submission[DATA_COMMON_NAME], 
+                                                            self.submission[MODEL_VERSION], cde_code, cde_version, self.log, self.mongo_dao)
+                    self.searched_sts = True
+                    if cde:
+                        if cde.get(CDE_PERMISSIVE_VALUES) is not None:
+                            if len(cde[CDE_PERMISSIVE_VALUES]) > 0:
+                                permissive_vals = cde[CDE_PERMISSIVE_VALUES]
+                            else:
+                                permissive_vals =  None #escape validation
+                    else:
+                        msg = CDE_NOT_FOUND
+                        self.not_found_cde = True
+                else: 
+                    if self.not_found_cde:
+                        msg = CDE_NOT_FOUND
+
+                       
         # strip white space if the value is string
         if permissive_vals and len(permissive_vals) > 0 and isinstance(permissive_vals[0], str):
             permissive_vals = [item.strip() for item in permissive_vals]
-        return permissive_vals
+        return permissive_vals, msg
 
 """util functions"""
 def check_permissive(value, permissive_vals, msg_prefix, prop_name, dao):

src/pv_puller.py

@@ -12,6 +12,8 @@
 FILE_DOWNLOAD_URL = "download_url"
 FILE_NAME = "name"
 FILE_TYPE = "type"
+CDE_PV_NAME = "permissibleValues"
+STS_FILE_URL = "https://raw.githubusercontent.com/CBIIT/crdc-datahub-terms/{}/{}_{}_sts.json"
 
 def pull_pv_lists(configs, mongo_dao):
     """
@@ -94,13 +96,9 @@ def extract_cde_from_file(self, download_url, cde_set, cde_record):
                 if cde_key in cde_set:
                     continue
                 cde_set.add(cde_key)
-                cde_long_name  = cde.get(CDE_FULL_NAME)
-                cde_record.append({
-                    CDE_FULL_NAME: cde_long_name,
-                    CDE_CODE: cde_code,
-                    CDE_VERSION: cde_version,
-                    CDE_PERMISSIVE_VALUES: extract_pv_list(cde.get('permissibleValues'))
-                })
+                cde_record.append(
+                    compose_cde_record(cde)
+                )
 
         except Exception as e:
             self.log.exception(e)
@@ -174,6 +172,70 @@ def get_pv_by_code_version(configs, log, data_common, prop_name, cde_code, cde_v
         UPDATED_AT: current_datetime(),
     }, msg
 
+def get_pv_by_datacommon_version_cde(tier, data_commons, data_model_version, cde_code, cde_version, log, mongo_dao):
+    """
+    extracts the CDE list by data_commons, data_model_version and finds matches based on the provided CDE code and version,
+    finally saves the CDE list into DB.
+
+    :param tier: The tier of the environment (e.g., 'dev', 'qa', 'prod').
+    :param data_commons: The name of the data commons.
+    :param data_model_version: The version of the data model.
+    :param cde_code: The code of the CDE to retrieve.
+    :param cde_version: The version of the CDE to retrieve.
+    :param log: Logger instance for logging information and errors.
+    :param mongo_dao: Data access object for MongoDB operations.
+    :returns: A dictionary containing the CDE full name, code, version, and permissive values, or None if the CDE is not found or an error occurs.
+    """
+    # construct the sts dump file url based on tier, data commons, and model version
+    sts_file_url = STS_FILE_URL.format(tier, data_commons.lower(), data_model_version)
+    try:
+        log.info(f"Extracting cde from {sts_file_url}")
+        api_client = APIInvoker({})
+        result = api_client.get_synonyms(sts_file_url)
+        if not result or len(result) == 0:
+            log.error(f"CDE dump file,{sts_file_url} is not found! ")
+            return None
+        cde_list  = [item for item in result if item.get(CDE_CODE) and item.get(CDE_CODE) != 'null'] 
+        if not cde_list or len(cde_list) == 0:
+            log.error(f"No cde found in {sts_file_url}")
+            return
+        
+        cde_set = set()
+        cde_records = []
+        return_cde = None
+        for item in cde_list:
+            code = item.get(CDE_CODE)
+            version = item.get(CDE_VERSION) if item.get(CDE_VERSION) and item.get(CDE_VERSION) != 'null' else None
+            cde_key = (code, version)
+            if cde_key in cde_set:
+                continue
+            cde_set.add(cde_key)
+            cde_record = compose_cde_record(item)
+            cde_records.append(
+                cde_record
+            )
+            if code  == cde_code and version == cde_version:
+                return_cde = cde_record
+        # save all extracted CDEs
+        mongo_dao.upsert_cde(cde_records)
+        # return matched CDE
+        return return_cde
+    except Exception as e:
+        log.exception(e)
+        log.exception(f"Failed to extract cde from {sts_file_url}")
+        return None
+    
+def compose_cde_record(cde_item):
+    """
+    compose cde record from cde dump file
+    """
+    cde_record = {
+        CDE_FULL_NAME: cde_item.get(CDE_FULL_NAME),
+        CDE_CODE: cde_item.get(CDE_CODE),
+        CDE_VERSION: cde_item.get(CDE_VERSION) if cde_item.get(CDE_VERSION) and cde_item.get(CDE_VERSION) != 'null' else None,
+        CDE_PERMISSIVE_VALUES: extract_pv_list(cde_item.get(CDE_PV_NAME))
+    }
+    return cde_record
 
 class SynonymPuller:
     """
@@ -238,9 +300,9 @@ def get_synonyms_by_datacommon_version(self, synonym_url, synonym_set):
                 self.log.info(f"Synonyms for {synonym_url} are not found! ")
                 return None
             # filter out empty synonyms
-            synonyms  = [item for item in result if item.get('permissibleValues') and item.get('permissibleValues')[0].get('synonyms')] 
+            synonyms  = [item for item in result if item.get(CDE_PV_NAME) and item.get(CDE_PV_NAME)[0].get('synonyms')] 
             for item in synonyms:
-                pv_list = item.get('permissibleValues')
+                pv_list = item.get(CDE_PV_NAME)
                 if pv_list:
                     for pv in pv_list:
                         value = pv.get('value')