Function _get_usage_trends_postgres

api/platform_hub/services.py:432–471

(
    organisations: QuerySet[Organisation],
    date_start: object,
)

def _get_usage_trends_postgres(
    organisations: QuerySet[Organisation],
    date_start: object,
) -> list[UsageTrendData]:
    env_ids = list(
        Environment.objects.filter(
            project__organisation__in=organisations,
        ).values_list("id", flat=True)
    )

    qs = (
        APIUsageBucket.objects.filter(
            environment_id__in=env_ids,
            bucket_size=analytics_constants.ANALYTICS_READ_BUCKET_SIZE,
            created_at__date__gt=date_start,
        )
        .values("created_at__date", "resource")
        .annotate(total=Sum("total_count"))
        .order_by("created_at__date")
    )

    daily: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
    for row in qs:
        date_str = row["created_at__date"].isoformat()
        resource = Resource(row["resource"])
        daily[date_str][resource.resource_name] = row["total"]

    results: list[UsageTrendData] = []
    for date_str in sorted(daily.keys()):
        data = daily[date_str]
        results.append(
            UsageTrendData(
                date=date_str,
                api_calls=sum(data.values()),
                flag_evaluations=data.get("flags", 0),
                identity_requests=data.get("identities", 0),
            )
        )

    return results


def _get_usage_trends_influx(