def parse_old()

in azure/Kqlmagic/parser.py [0:0]

• 81 lines of code
• 26 McCabe index (conditional complexity)

    def parse_old(cls, line:str, cell:str, config:Configurable, engines:List[Engine], user_ns:Dict[str,Any])->List[Dict[str,Any]]:
        """Separate input into (connection info, KQL statements, options)"""

        is_cell = cell is not None
        cell = f"{line}\n{cell or ''}"
        cell = cell.strip()
        parsed_queries = []
        cell, command = cls._parse_kql_command(cell, user_ns)
        command_name = command.get("command")
        if command_name is not None and command_name != "submit":
            cell_rest, options = cls._parse_kql_options(cell.strip(), is_cell, config, user_ns)
            cell_rest = cls._update_kql_command_params(command, cell_rest, user_ns)
            cls._validate_kql_command_params(command)
            if cell_rest: 
                raise ValueError(f"command --{command_name} has too many parameters")
            parsed_queries.append({"connection_string": "", "query": "", "options": options, "command": command})
            return parsed_queries

        # split to max 2 parts. First part, parts[0], is the first string.
        # parts = [part.strip() for part in cell.split(None, 1)]
        parts = split_lex(cell)
        
        # print(parts)
        if not parts:
            kql, options = cls._parse_kql_options("", is_cell, config, user_ns)
            parsed_queries.append({"connection_string": "", "query": kql, "options": options, "command": {}})
            return parsed_queries

        #
        # replace substring of the form $name or ${name}, in windows also %name% if found in env variabes
        #

        connection_string = None

        conn_str = parts[0].strip()
        if not conn_str.startswith(('-', '+')):
            _was_quoted, conn_str = strip_if_quoted(conn_str)

            #
            # connection taken from a section in  dsn file (file name have to be define in config.dsn_filename or specified as a parameter)
            #
            if is_collection(conn_str, "["):
                section = conn_str[1:-1].strip()

                # parse to get flag, for the case that the file nema is specified in the options
                code = cell[len(parts[0]):]
                kql, options = cls._parse_kql_options(code, is_cell, config, user_ns)

                parser = CP.ConfigParser()
                dsn_filename = adjust_path(options.get("dsn_filename", config.dsn_filename))
                parser.read(dsn_filename)
                cfg_dict = dict(parser.items(section))

                cfg_dict_lower = {k.lower().replace("_", "").replace("-", ""): v for (k, v) in cfg_dict.items()}
                for e in engines:
                    if e.get_mandatory_key() in cfg_dict_lower.keys():
                        all_keys = set(itertools.chain(*e.get_valid_keys_combinations()))
                        connection_kv = [f"{k}='{v}'" for k, v in cfg_dict_lower.items() if v and k in all_keys]
                        connection_string = f"{e.get_uri_schema_name()}://{';'.join(connection_kv)}"
                        break

            #
            # connection specified starting with one of the supported prefixes
            #
            elif "://" in conn_str:
                sub_parts = conn_str.strip().split("://", 1)
                if (len(sub_parts) == 2 and sub_parts[0].lower().replace("_", "").replace("-", "") in list(itertools.chain(*[e._ALT_URI_SCHEMA_NAMES for e in engines]))):
                    connection_string = conn_str
            #
            # connection specified as database@cluster
            #
            elif "@" in conn_str and "|" not in conn_str and "'" not in conn_str and '"' not in conn_str:
                connection_string = conn_str
        #
        # connection not specified, override default
        #
        if connection_string is None:
            connection_string = ""
            code = cell
        else:
            code = cell[len(parts[0]):]

        #
        # split string to queries
        #
        suppress_all_results = False
        queries:List[str] = []
        if is_cell:
            queryLines:List[str] = []
            last_line:str = None
            for last_line in code.splitlines(True):
                # note: splitlines don't remove the \n suffix, each line endswith \n 
                if last_line.isspace():
                    if len(queryLines) > 0:
                        queries.append("".join(queryLines))
                        queryLines = []
                else:
                    queryLines.append(last_line)

            if len(queryLines) > 0:
                queries.append("".join(queryLines))

            if len(queries) > 0:
                last_query = queries[-1].strip()
                if last_query == ";":
                    suppress_all_results = True
                    queries = queries[:-1]

            if len(queries) == 0:
                queries.append("")
        else:
            queries.append(code.strip())

        #
        # parse code to kql and options
        #
        for query in queries:
            kql, options = cls._parse_kql_options(query.strip(), is_cell, config, user_ns)
            kql = options.pop("query", None) or kql
            connection_string = options.pop("conn", None) or connection_string
            if suppress_all_results:
                options["suppress_results"] = True
            parsed_queries.append({"connection_string": connection_string.strip(), "query": kql, "options": options, "command": {}})

        return parsed_queries