def cache_key()

in superset/common/query_object.py [0:0]

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

    def cache_key(self, **extra: Any) -> str:  # noqa: C901
        """
        The cache key is made out of the key/values from to_dict(), plus any
        other key/values in `extra`
        We remove datetime bounds that are hard values, and replace them with
        the use-provided inputs to bounds, which may be time-relative (as in
        "5 days ago" or "now").
        """
        cache_dict = self.to_dict()
        cache_dict.update(extra)

        # TODO: the below KVs can all be cleaned up and moved to `to_dict()` at some
        #  predetermined point in time when orgs are aware that the previously
        #  cached results will be invalidated.
        if not self.apply_fetch_values_predicate:
            del cache_dict["apply_fetch_values_predicate"]
        if self.datasource:
            cache_dict["datasource"] = self.datasource.uid
        if self.result_type:
            cache_dict["result_type"] = self.result_type
        if self.time_range:
            cache_dict["time_range"] = self.time_range
        if self.post_processing:
            cache_dict["post_processing"] = self.post_processing
        if self.time_offsets:
            cache_dict["time_offsets"] = self.time_offsets

        for k in ["from_dttm", "to_dttm"]:
            del cache_dict[k]

        annotation_fields = [
            "annotationType",
            "descriptionColumns",
            "intervalEndColumn",
            "name",
            "overrides",
            "sourceType",
            "timeColumn",
            "titleColumn",
            "value",
        ]
        annotation_layers = [
            {field: layer[field] for field in annotation_fields if field in layer}
            for layer in self.annotation_layers
        ]
        # only add to key if there are annotations present that affect the payload
        if annotation_layers:
            cache_dict["annotation_layers"] = annotation_layers

        # Add an impersonation key to cache if impersonation is enabled on the db
        # or if the CACHE_QUERY_BY_USER flag is on
        try:
            database = self.datasource.database  # type: ignore
            if (
                feature_flag_manager.is_feature_enabled("CACHE_IMPERSONATION")
                and database.impersonate_user
            ) or feature_flag_manager.is_feature_enabled("CACHE_QUERY_BY_USER"):
                if key := database.db_engine_spec.get_impersonation_key(
                    getattr(g, "user", None)
                ):
                    logger.debug(
                        "Adding impersonation key to QueryObject cache dict: %s", key
                    )

                    cache_dict["impersonation_key"] = key
        except AttributeError:
            # datasource or database do not exist
            pass

        return md5_sha_from_dict(cache_dict, default=json_int_dttm_ser, ignore_nan=True)