def _parse_one_section()

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

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

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

        cell, command = cls._parse_kql_command(section, 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_query = {"connection_string": "", "query": "", "options": options, "command": command}
            return parsed_query

        # 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_query = {"connection_string": "", "query": kql, "options": options, "command": {}}
            return parsed_query

        #
        # 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]):]


        #
        # parse code to kql and options
        #

        kql, options = cls._parse_kql_options(code.strip(), is_cell, config, user_ns)
        kql = options.pop("query", None) or kql
        connection_string = options.pop("conn", None) or connection_string

        parsed_query = {"connection_string": connection_string.strip(), "query": kql, "options": options, "command": {}}

        return parsed_query