def has_extra_cache_key_calls()

in superset/connectors/sqla/models.py [0:0]

• 39 lines of code
• 23 McCabe index (conditional complexity)

    def has_extra_cache_key_calls(self, query_obj: QueryObjectDict) -> bool:  # noqa: C901
        """
        Detects the presence of calls to `ExtraCache` methods in items in query_obj that
        can be templated. If any are present, the query must be evaluated to extract
        additional keys for the cache key. This method is needed to avoid executing the
        template code unnecessarily, as it may contain expensive calls, e.g. to extract
        the latest partition of a database.

        :param query_obj: query object to analyze
        :return: True if there are call(s) to an `ExtraCache` method, False otherwise
        """
        templatable_statements: list[str] = []
        if self.sql:
            templatable_statements.append(self.sql)
        if self.fetch_values_predicate:
            templatable_statements.append(self.fetch_values_predicate)
        extras = query_obj.get("extras", {})
        if "where" in extras:
            templatable_statements.append(extras["where"])
        if "having" in extras:
            templatable_statements.append(extras["having"])
        if columns := query_obj.get("columns"):
            calculated_columns: dict[str, Any] = {
                c.column_name: c.expression for c in self.columns if c.expression
            }
            for column_ in columns:
                if utils.is_adhoc_column(column_):
                    templatable_statements.append(column_["sqlExpression"])
                elif isinstance(column_, str) and column_ in calculated_columns:
                    templatable_statements.append(calculated_columns[column_])
        if metrics := query_obj.get("metrics"):
            metrics_by_name: dict[str, Any] = {
                m.metric_name: m.expression for m in self.metrics
            }
            for metric in metrics:
                if utils.is_adhoc_metric(metric) and (
                    sql := metric.get("sqlExpression")
                ):
                    templatable_statements.append(sql)
                elif isinstance(metric, str) and metric in metrics_by_name:
                    templatable_statements.append(metrics_by_name[metric])
        if self.is_rls_supported:
            templatable_statements += [
                f.clause for f in security_manager.get_rls_filters(self)
            ]
        for statement in templatable_statements:
            if ExtraCache.regex.search(statement):
                return True
        return False