from dataclasses import dataclass import sqlparse from models.ExportType import ExportType @dataclass class QueryParameters: _input_path: str = None _query: str = None _export_type: ExportType = None _query_parameters: dict = None _title: str = None def __init__(self, title: str, input_path: str = None, query: str = None, export_type: ExportType = None, query_parameters: dict = None): if input_path is None and query is None: raise ValueError("Either input_path or query must be provided") if input_path is not None and query is not None: raise ValueError("Cannot have both input_path and query") self._input_path = input_path self._query = query self._export_type = export_type self._query_parameters = query_parameters self._title = title def _split_query_to_singular_queries(self, query: str) -> list[str]: query = sqlparse.format(query, strip_comments=True) return [query for query in sqlparse.split(query)] def _read_and_sql_file_and_strip_for_comments(self, filename: str) -> list[str]: query: str with open(filename, 'r') as f: query = f.read() return self._split_query_to_singular_queries(query) def get_queries(self) -> list[str]: if self._input_path is not None: return self._read_and_sql_file_and_strip_for_comments(self._input_path) if self._query is not None: return self._split_query_to_singular_queries(self._query) raise Exception(f'No query provided') @property def query_parameters(self) -> dict[str, str]: return self._query_parameters @property def output_path(self) -> str: return self._output_path @property def export_type(self) -> ExportType: return self._export_type @property def input_path(self) -> str: return self._input_path @property def title(self) -> str: return self._title