Merge pull request #945 from Sage-Bionetworks/develop-table-schema-fix

Fixing parsing of table schema parameters on table replace

Author: GiaJordan

schematic/store/synapse.py

@@ -138,8 +138,11 @@ def wrapper(*args, **kwargs):
                 return method(*args, **kwargs)
             except(SynapseHTTPError) as ex:
                 str_message = str(ex).replace("\n","")
-                logging.warning(str_message)
-                return None
+                if 'missing' in str_message:
+                    logging.warning(str_message)
+                    return None
+                else:
+                    raise ex
         return wrapper
 
 
@@ -1103,9 +1106,8 @@ def associateMetadataWithFiles(
         # Load manifest to synapse as a CSV File
         manifest_synapse_file_id = self.uplodad_manifest_file(manifest, metadataManifestPath, datasetId, restrict_manifest, component_name = component_name)
 
-        # Get annotations for the file manifest.
+        # Set annotations for the file manifest.
         manifest_annotations = self.format_manifest_annotations(manifest, manifest_synapse_file_id)
-        
         self.syn.set_annotations(manifest_annotations)
 
         logger.info("Associated manifest file with dataset on Synapse.")
@@ -1122,7 +1124,7 @@ def associateMetadataWithFiles(
                 restrict = restrict_manifest
                 )
 
-            # Get annotations for the table manifest
+            # Set annotations for the table manifest
             manifest_annotations = self.format_manifest_annotations(manifest, manifest_synapse_table_id)
             self.syn.set_annotations(manifest_annotations)
 
@@ -1350,6 +1352,18 @@ def getDatasetAnnotationsBatch(
 
         return table
 
+    def _get_table_schema_by_cname(self, table_schema):
+
+        # assume no duplicate column names in the table
+        table_schema_by_cname = {}
+
+        for col_record in table_schema:
+
+            #TODO clean up dictionary for compactness (e.g. remove redundant 'name' key)
+            table_schema_by_cname[col_record["name"]] = col_record
+
+        return table_schema_by_cname
+
     def make_synapse_table(self, 
             table_to_load: pd.DataFrame, 
             dataset_id: str, table_name: str, 
@@ -1392,34 +1406,76 @@ def make_synapse_table(self,
             # locate its position in the old and new table.
             if manipulation == 'update':
                 table_to_load = update_df(existing_table, table_to_load, update_col)
-            
+                # store table with existing etag data and impose restrictions as appropriate
+                self.syn.store(Table(existingTableId, table_to_load, etag = existing_results.etag), isRestricted = restrict)
+
             elif manipulation == 'replace':
                 # remove rows
                 self.syn.delete(existing_results)
-
                 # wait for row deletion to finish on synapse before getting empty table
-                sleep(5)
+                sleep(1)
+                
                 # removes all current columns
                 current_table = self.syn.get(existingTableId)
                 current_columns = self.syn.getTableColumns(current_table)
                 for col in current_columns:
                     current_table.removeColumn(col)
 
-                # adds new columns to schema
-                new_columns = as_table_columns(table_to_load)
-                for col in new_columns:
-                    current_table.addColumn(col)
-                self.syn.store(current_table, isRestricted = restrict)
-
-            # store table with existing etag data and impose restrictions as appropriate
-            self.syn.store(Table(existingTableId, table_to_load, etag = existing_results.etag), isRestricted = restrict)
+                if not table_name:
+                    table_name = datasetName + 'table'
+                
+                # Process columns according to manifest entries
+                table_schema_by_cname = self._get_table_schema_by_cname(column_type_dictionary) 
+                datasetParentProject = self.getDatasetProject(dataset_id)
+                if specify_schema:
+                    if column_type_dictionary == {}:
+                        logger.error("Did not provide a column_type_dictionary.")
+                    #create list of columns:
+                    cols = []
+                    
+                    for col in table_to_load.columns:
+                        
+                        if col in table_schema_by_cname:
+                            col_type = table_schema_by_cname[col]['columnType']
+                            max_size = table_schema_by_cname[col]['maximumSize']
+                            max_list_len = 250
+                            if max_size and max_list_len:
+                                cols.append(Column(name=col, columnType=col_type, 
+                                    maximumSize=max_size, maximumListLength=max_list_len))
+                            elif max_size:
+                                cols.append(Column(name=col, columnType=col_type, 
+                                    maximumSize=max_size))
+                            else:
+                                cols.append(Column(name=col, columnType=col_type))
+                        else:
+                            
+                            #TODO add warning that the given col was not found and it's max size is set to 100
+                            cols.append(Column(name=col, columnType='STRING', maximumSize=100))
+                    
+                    # adds new columns to schema
+                    for col in cols:
+                        current_table.addColumn(col)
+                    self.syn.store(current_table, isRestricted = restrict)
+
+                    # wait for synapse store to finish
+                    sleep(1)
+
+                    # build schema and table from columns and store with necessary restrictions
+                    schema = Schema(name=table_name, columns=cols, parent=datasetParentProject)
+                    schema.id = existingTableId
+                    table = Table(schema, table_to_load, etag = existing_results.etag)
+                    table = self.syn.store(table, isRestricted = restrict)
+                else:
+                    logging.error("Must specify a schema for table replacements")
 
             # remove system metadata from manifest
             existing_table.drop(columns = ['ROW_ID', 'ROW_VERSION'], inplace = True)
             return existingTableId
         else:
             datasetEntity = self.syn.get(dataset_id, downloadFile = False)
             datasetName = datasetEntity.name
+            table_schema_by_cname = self._get_table_schema_by_cname(column_type_dictionary) 
+
             if not table_name:
                 table_name = datasetName + 'table'
             datasetParentProject = self.getDatasetProject(dataset_id)
@@ -1429,10 +1485,10 @@ def make_synapse_table(self,
                 #create list of columns:
                 cols = []
                 for col in table_to_load.columns:
-                    if col in column_type_dictionary:
-                        col_type = column_type_dictionary[col]['column_type']
-                        max_size = column_type_dictionary[col]['maximum_size']
-                        max_list_len = column_type_dictionary[col]['maximum_list_length']
+                    if col in table_schema_by_cname:
+                        col_type = table_schema_by_cname[col]['columnType']
+                        max_size = table_schema_by_cname[col]['maximumSize']
+                        max_list_len = 250
                         if max_size and max_list_len:
                             cols.append(Column(name=col, columnType=col_type, 
                                 maximumSize=max_size, maximumListLength=max_list_len))
@@ -1442,7 +1498,8 @@ def make_synapse_table(self,
                         else:
                             cols.append(Column(name=col, columnType=col_type))
                     else:
-                        cols.append(Column(name=col, columnType='STRING', maximumSize=500))
+                        #TODO add warning that the given col was not found and it's max size is set to 100
+                        cols.append(Column(name=col, columnType='STRING', maximumSize=100))
                 schema = Schema(name=table_name, columns=cols, parent=datasetParentProject)
                 table = Table(schema, table_to_load)
                 table = self.syn.store(table, isRestricted = restrict)