fix validate-dtable-db-cell-value

Author: AlexCXC

dtable_events/common_dataset/common_dataset_sync_utils.py

@@ -698,196 +698,6 @@ def delete_dst_rows(dst_dtable_uuid, dst_table_name, to_be_deleted_row_ids, dst_
 
 
 def import_sync_CDS(context):
-    """
-    import or sync common dataset
-    please check all resources in context before call this function
-
-    Steps:
-        1. create or update dst columns
-        2. fetch src rows, (find rows to be updated and rows to be appended, update and append them), step by step
-        3. fetch dst rows, (find rows to be deleted, delete them), step by step
-    """
-    src_dtable_uuid = context.get('src_dtable_uuid')
-    dst_dtable_uuid = context.get('dst_dtable_uuid')
-
-    src_table_name = context.get('src_table_name')
-    src_view_name = context.get('src_view_name')
-    src_view_type = context.get('src_view_type', 'table')
-    src_columns = context.get('src_columns')
-    src_enable_archive = context.get('src_enable_archive', False)
-
-    dst_table_id = context.get('dst_table_id')
-    dst_table_name = context.get('dst_table_name')
-    dst_columns = context.get('dst_columns')
-
-    operator = context.get('operator')
-    lang = context.get('lang', 'en')
-
-    to_archive = context.get('to_archive', False)
-
-    src_dtable_server_api = DTableServerAPI(operator, src_dtable_uuid, dtable_server_url)
-    src_dtable_db_api = DTableDBAPI(operator, src_dtable_uuid, INNER_DTABLE_DB_URL)
-    dst_dtable_server_api = DTableServerAPI(operator, dst_dtable_uuid, dtable_server_url)
-    dst_dtable_db_api = DTableDBAPI(operator, dst_dtable_uuid, INNER_DTABLE_DB_URL)
-
-    server_only = not (to_archive and src_enable_archive and src_view_type == 'archive')
-    is_sync = bool(dst_table_id)
-    logger.debug('to_archive: %s src_enable_archive: %s src_view_type: %s', to_archive, src_enable_archive, src_view_type)
-
-    src_column_keys_set = {col['key'] for col in src_columns}
-
-    # fetch src rows, find existed rows, not existed rows, update/append rows, step by step
-    start, step = 0, 10000
-    src_row_ids_set = set()
-    to_be_updated_columns, to_be_appended_columns = [], []
-    final_columns = []
-    while True:
-        logger.debug('update/append start: %s step: %s', start, step)
-        if server_only and (start + step) > SRC_ROWS_LIMIT:
-            step = SRC_ROWS_LIMIT - start
-        try:
-            res_json = src_dtable_server_api.internal_view_rows(src_table_name, src_view_name, use_dtable_db=True, server_only=server_only, start=start, limit=step)
-            step_src_rows = res_json.get('rows', [])
-            src_view_metadata = res_json.get('metadata')
-        except Exception as e:
-            logger.error('request src_dtable: %s view-rows error: %s', src_dtable_uuid, e)
-            return {
-                'dst_table_id': None,
-                'error_msg': 'fetch view rows error',
-                'task_status_code': 500
-            }
-        if start == 0: ## generate columns from the columns(archive_metadata) returned from SQL query in internal_view_rows
-            sync_columns = [col for col in src_view_metadata if col['key'] in src_column_keys_set]
-            to_be_updated_columns, to_be_appended_columns, error = generate_synced_columns(sync_columns, dst_columns=dst_columns)
-            if error:
-                return {
-                    'dst_table_id': None,
-                    'error_type': 'generate_synced_columns_error',
-                    'error_msg': str(error),  # generally, this error is caused by client
-                    'task_status_code': 400
-                }
-            final_columns = (to_be_updated_columns or []) + (to_be_appended_columns or [])
-            ### create or update dst columns
-            dst_table_id, error_resp = create_dst_table_or_update_columns(dst_dtable_uuid, dst_table_id, dst_table_name, to_be_appended_columns, to_be_updated_columns, dst_dtable_server_api, lang)
-            if error_resp:
-                return error_resp
-        row_ids = []
-        step_rows_dict = {}
-        for row in step_src_rows:
-            if row['_id'] in src_row_ids_set:
-                continue
-            row_ids.append(row['_id'])
-            src_row_ids_set.add(row['_id'])
-            step_rows_dict[row['_id']] = row
-        if not row_ids:
-            if not step_src_rows or len(step_src_rows) < step or (server_only and (start + step) >= SRC_ROWS_LIMIT):
-                break
-            start += step
-            continue
-        ## find to-be-appended-rows to-be-updated-rows
-        step_dst_rows = None
-        if dst_table_id:
-            sql = "SELECT _id, %(dst_columns)s FROM `%(dst_table)s` WHERE _id IN %(row_ids)s LIMIT %(rows_count)s" % {
-                'dst_table': dst_table_name,
-                'dst_columns': ', '.join(["`%s`" % col['name'] for col in final_columns]),
-                'row_ids': '(%s)' % ', '.join(["'%s'" % row_id for row_id in row_ids]),
-                'rows_count': len(row_ids)
-            }
-            try:
-                step_dst_rows = dst_dtable_db_api.query(sql, convert=False, server_only=(not to_archive))
-            except Exception as e:
-                logger.error('find to-be-updated-rows error: %s', e)
-                return {
-                    'dst_table_id': None,
-                    'error_msg': 'find to-be-updated-rows error',
-                    'task_status_code': 500
-                }
-        logger.debug('step_dst_rows: %s', len(step_dst_rows))
-        filtered_step_src_rows = [step_rows_dict[row_id] for row_id in row_ids]
-        to_be_updated_rows, to_be_appended_rows, _ = generate_synced_rows(filtered_step_src_rows, src_columns, final_columns, dst_rows=step_dst_rows, to_archive=to_archive)
-        logger.debug('to_be_updated_rows: %s to_be_appended_rows: %s', len(to_be_updated_rows), len(to_be_appended_rows))
-        ## append
-        if to_be_appended_rows:
-            error_resp = append_dst_rows(dst_dtable_uuid, dst_table_name, to_be_appended_rows, dst_dtable_db_api, dst_dtable_server_api, to_archive)
-            if error_resp:
-                return error_resp
-        ## update
-        if to_be_updated_rows:
-            error_resp = update_dst_rows(dst_dtable_uuid, dst_table_name, to_be_updated_rows, dst_dtable_db_api, dst_dtable_server_api, to_archive)
-            if error_resp:
-                return error_resp
-        ## judge whether break
-        if not step_src_rows or len(step_src_rows) < step or (server_only and (start + step) >= SRC_ROWS_LIMIT):
-            break
-        start += step
-
-    # fetch dst rows, find useless rows, delete rows, step by step
-    dst_row_ids_set = set()
-    start, step = 0, 10000
-    ## generate src view WHERE clause
-    try:
-        src_metadata = src_dtable_server_api.get_metadata()
-    except Exception as e:
-        logger.error('request src dtable: %s table: %s metadata error: %s', src_dtable_uuid, src_table_name, e)
-        return {
-            'dst_table_id': None,
-            'error_msg': 'request src table metadata error',
-            'task_status_code': 500
-        }
-    src_table = [table for table in src_metadata['tables'] if table['name'] == src_table_name][0]
-    src_view = [view for view in src_table['views'] if view['name'] == src_view_name][0]
-    filters = src_view.get('filters', [])
-    filter_conjunction = src_view.get('filter_conjunction', 'And')
-    filter_conditions = {
-        'filters': filters,
-        'filter_conjunction': filter_conjunction
-    }
-    sql_generator = BaseSQLGenerator(src_table_name, src_table['columns'], filter_conditions=filter_conditions)
-    filter_clause = sql_generator._filter2sql()
-    ## delete useless rows step by step
-    while is_sync and True:
-        logger.debug('delete start: %s step: %s', start, step)
-        sql = "SELECT _id FROM `%(dst_table_name)s` LIMIT %(start)s, %(limit)s" % {
-            'dst_table_name': dst_table_name,
-            'start': start,
-            'limit': step
-        }
-        rows = dst_dtable_db_api.query(sql, convert=False, server_only=(not to_archive))
-        query_row_ids_set = set()
-        for row in rows:
-            if row['_id'] in dst_row_ids_set:
-                continue
-            if row['_id'] in src_row_ids_set:
-                continue
-            query_row_ids_set.add(row['_id'])
-            dst_row_ids_set.add(row['_id'])
-        if not query_row_ids_set:
-            if len(rows) < step:
-                break
-            start += step
-            continue
-        sql = "SELECT _id FROM `%(src_table_name)s` WHERE _id IN %(row_ids)s %(view_filter_clause)s LIMIT %(rows_count)s" % {
-            'src_table_name': src_table_name,
-            'row_ids': '(%s)' % ', '.join(["'%s'" % row_id for row_id in query_row_ids_set]),
-            'rows_count': len(query_row_ids_set),
-            'view_filter_clause': 'AND (%s)' % filter_clause[filter_clause.find('WHERE') + len('WHERE'):] if filter_clause else ''
-        }
-        existed_rows = src_dtable_db_api.query(sql, convert=False, server_only=server_only)
-        to_be_deleted_row_ids_set = query_row_ids_set - {row['_id'] for row in existed_rows}
-        if to_be_deleted_row_ids_set:
-            delete_dst_rows(dst_dtable_uuid, dst_table_name, list(to_be_deleted_row_ids_set), dst_dtable_db_api, dst_dtable_server_api, to_archive)
-        if len(rows) < step:
-                break
-        start += step
-
-    return {
-        'dst_table_id': dst_table_id,
-        'error_msg': None,
-        'task_status_code': 200
-    }
-
-
-def new_import_sync_CDS(context):
     """
     fetch src/dst rows id, find need append/update/delete rows
     """
@@ -918,8 +728,6 @@ def new_import_sync_CDS(context):
     is_sync = bool(dst_table_id)
     logger.debug('to_archive: %s src_enable_archive: %s src_view_type: %s', to_archive, src_enable_archive, src_view_type)
 
-    src_column_keys_set = {col['key'] for col in src_columns}
-
     # fetch create dst table or update dst table columns
     # fetch all src view rows id, S
     # fetch all dst table rows id, D

dtable_events/common_dataset/dtable_db_cell_validators.py

@@ -49,6 +49,8 @@ def validate(self, value):
         if value is None:
             return None
         if isinstance(value, list):
+            if value and not isinstance(value[0], str):
+                return None
             return [v for v in value if isinstance(v, str)]
         return None
 
@@ -91,7 +93,7 @@ def validate(self, value):
 class MultipleSelectValidator(BaseValidator):
 
     def validate(self, value):
-        if not isinstance(value, list):
+        if not isinstance(value, list) or (value and not isinstance(value[0], str)):
             return None
         column_data = self.column.get('data') or {}
         options = column_data.get('options') or []
@@ -115,7 +117,7 @@ def validate(self, value):
 class FileValidator(BaseValidator):
 
     def validate(self, value):
-        if not isinstance(value, list):
+        if not isinstance(value, list) or (value and not isinstance(value[0], dict)):
             return None
         real_values = []
         for v in value:
@@ -129,7 +131,7 @@ def validate(self, value):
 class CollaboratorValidator(BaseValidator):
 
     def validate(self, value):
-        if not isinstance(value, list):
+        if not isinstance(value, list) or (value and not isinstance(value[0], str)):
             return None
         return [v for v in value if isinstance(v, str)]
 

dtable_events/dtable_io/import_sync_common_dataset.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 
 import requests
-from dtable_events.common_dataset.common_dataset_sync_utils import import_sync_CDS, new_import_sync_CDS
+from dtable_events.common_dataset.common_dataset_sync_utils import import_sync_CDS
 from dtable_events.db import init_db_session_class
 from dtable_events.dtable_io import dtable_io_logger
 from dtable_events.utils import uuid_str_to_32_chars, get_inner_dtable_server_url
@@ -75,7 +75,7 @@ def sync_common_dataset(context, config):
             return
 
     try:
-        result = new_import_sync_CDS({
+        result = import_sync_CDS({
             'src_dtable_uuid': src_dtable_uuid,
             'dst_dtable_uuid': dst_dtable_uuid,
             'src_table_name': src_table_name,
@@ -155,7 +155,7 @@ def import_common_dataset(context, config):
     dataset_id = context.get('dataset_id')
 
     try:
-        result = new_import_sync_CDS({
+        result = import_sync_CDS({
             'src_dtable_uuid': src_dtable_uuid,
             'dst_dtable_uuid': dst_dtable_uuid,
             'src_table_name': src_table_name,