def _execute_query()

in msticpy/data/data_providers.py [0:0]

• 41 lines of code
• 17 McCabe index (conditional complexity)

    def _execute_query(self, *args, **kwargs) -> Union[pd.DataFrame, Any]:
        if not self._query_provider.loaded:
            raise ValueError("Provider is not loaded.")
        if not self._query_provider.connected:
            raise ValueError(
                "No connection to a data source.",
                "Please call connect(connection_str) and retry.",
            )
        query_name = kwargs.pop("query_name")
        family = kwargs.pop("query_path")

        query_source = self.query_store.get_query(
            query_path=family, query_name=query_name
        )
        if "help" in args or "?" in args:
            query_source.help()
            return None

        params, missing = extract_query_params(query_source, *args, **kwargs)
        self._check_for_time_params(params, missing)
        if missing:
            query_source.help()
            raise ValueError(f"No values found for these parameters: {missing}")

        split_by = kwargs.pop("split_query_by", None)
        if split_by:
            split_result = self._exec_split_query(
                split_by=split_by,
                query_source=query_source,
                query_params=params,
                args=args,
                **kwargs,
            )
            if split_result is not None:
                return split_result
            # if split queries could not be created, fall back to default
        query_str = query_source.create_query(
            formatters=self._query_provider.formatters, **params
        )
        # This looks for any of the "print query" debug args in args or kwargs
        if any(db_arg for db_arg in _DB_QUERY_FLAGS if db_arg in args) or any(
            db_arg for db_arg in _DB_QUERY_FLAGS if kwargs.get(db_arg, False)
        ):
            return query_str

        # Handle any query options passed
        query_options = self._get_query_options(params, kwargs)
        return self.exec_query(query_str, query_source=query_source, **query_options)