Untitled
unknown
plain_text
a year ago
3.4 kB
5
Indexable
""" Check if script selects from sandbox(_analysts) schema """ import logging import re import yaml from scripts.validator.linters.dwh.common_dwh_linter_factory import common_dwh_linter_factory from simple_ddl_parser import DDLParser path = "../../../../../dwh/dm/dm_detail_ad" def sandbox_check(files: dict): ddl_pars = DDLParser(files["ddl"]).run(output_mode="hql") # Проверка, что есть DDL if not ddl_pars: logging.error("Не удалось распарсить DDL файл") return False ddl_table_name = ddl_pars[0].get('table_name',"") # Проверка, что название таблицы не пусто if not ddl_table_name: logging.error("Не удалось получить название таблицы из DDL") return False yml_pars = yaml.safe_load(files["specification"]) # Проверка, что выполняется парсинг yml if type(yml_pars) is not dict: logging.error("Не удалось распарсить YML файл") return False yml_table_name = yml_pars.get('table-name', "") # Проверка, что есть название таблицы if not yml_table_name: logging.error("Не удалось получить название таблицы из YML") return False if ddl_table_name != yml_table_name: logging.error("Наименование таблиц в DDL и YML не совпали") return False logging.info("Наименование таблиц совпали") # TODO: нужно добавить проверку по совпадению длины (лист ddl=лист yml) colum_index = 0 for ddl_column in ddl_pars[0]['columns']: #logging.info(ddl_column['name']) #a = ddl_column['name'] #b = yml_pars['structure'][colum_index]['column-name'] yml_column = yml_pars['structure'][colum_index] if ddl_column['name'] != yml_column['column-name']: logging.error(f"Наименования колонок в DDL и YML различаются в позиици {colum_index}") return False ddl_column_type = str.upper(ddl_column['type']) yml_column_type = str.upper(yml_column['data-type']).split('(')[0] if ddl_column_type != yml_column_type: logging.error(f"Типы колонок в DDL и YML различаются в позиици {colum_index}") return False #if ddl_column поверить что none не пусто ( и наоборрот ) #ddl_column_size = число #yml_column_size = конвертнуть строку в число и применить регулярку #if ddl_column_size != yml_column_size: # logging.error(f"Размерность колонок в DDL и YML различаются в позиици {colum_index}") # return False colum_index += 1 logging.info("Данные колонок совпали") # проверка колонок в yml #for yml_column in yml_pars['structure']: # logging.info(yml_column['column-name']) return True common_check = common_dwh_linter_factory([sandbox_check]) common_check.set_path(path) common_check.local_test()
Editor is loading...
Leave a Comment