Class Trends

posthog/queries/trends/trends.py:34–254

class Trends(TrendsTotalVolume, Lifecycle, TrendsFormula):
    def _get_sql_for_entity(self, filter: Filter, team: Team, entity: Entity) -> Tuple[str, str, Dict, Callable]:
        if filter.breakdown and filter.display not in NON_BREAKDOWN_DISPLAY_TYPES:
            query_type = "trends_breakdown"
            sql, params, parse_function = TrendsBreakdown(
                entity, filter, team, using_person_on_events=team.person_on_events_querying_enabled
            ).get_query()
        elif filter.shown_as == TRENDS_LIFECYCLE:
            query_type = "trends_lifecycle"
            sql, params, parse_function = self._format_lifecycle_query(entity, filter, team)
        else:
            query_type = "trends_total_volume"
            sql, params, parse_function = self._total_volume_query(entity, filter, team)

        return query_type, sql, params, parse_function

    # Use cached result even on refresh if team has strict caching enabled
    def get_cached_result(self, filter: Filter, team: Team) -> Optional[List[Dict[str, Any]]]:

        if not team.strict_caching_enabled or filter.breakdown or filter.display != TRENDS_LINEAR:
            return None

        cache_key = generate_cache_key(f"{filter.toJSON()}_{team.pk}")
        cached_result_package = get_safe_cache(cache_key)
        cached_result = (
            cached_result_package.get("result")
            if cached_result_package and isinstance(cached_result_package, dict)
            else None
        )

        if not cached_result:
            return None

        _is_present = self.is_present_timerange(cached_result, filter, team)

        return cached_result if _is_present else None

    # Determine if the current timerange is present in the cache
    def is_present_timerange(self, cached_result: List[Dict[str, Any]], filter: Filter, team: Team) -> bool:
        if (
            len(cached_result) > 0
            and cached_result[0].get("days")
            and cached_result[0].get("data")
            and len(cached_result[0]["days"]) > 0
            and len(cached_result[0]["days"]) == len(cached_result[0]["data"])
        ):
            latest_date = cached_result[0]["days"][len(cached_result[0]["days"]) - 1]

            parsed_latest_date = parser.parse(latest_date)
            parsed_latest_date = parsed_latest_date.replace(tzinfo=pytz.timezone(team.timezone))
            _is_present = is_filter_date_present(filter, parsed_latest_date)
        else:
            _is_present = False

        return _is_present

    # Use a condensed filter if a cached result exists in the current timerange
    def adjusted_filter(self, filter: Filter, team: Team) -> Tuple[Filter, Optional[Dict[str, Any]]]: