Get per-environment API usage from Postgres for the given number of days.
(
environment_ids: list[int],
days: int = 30,
)
| 128 | |
| 129 | |
| 130 | def _get_api_usage_for_envs_postgres( |
| 131 | environment_ids: list[int], |
| 132 | days: int = 30, |
| 133 | ) -> dict[int, dict[str, int]]: |
| 134 | """Get per-environment API usage from Postgres for the given number of days.""" |
| 135 | date_start = timezone.now() - timedelta(days=days) |
| 136 | qs = ( |
| 137 | APIUsageBucket.objects.filter( |
| 138 | environment_id__in=environment_ids, |
| 139 | bucket_size=analytics_constants.ANALYTICS_READ_BUCKET_SIZE, |
| 140 | created_at__date__gt=date_start.date(), |
| 141 | ) |
| 142 | .values("environment_id", "resource") |
| 143 | .annotate(total=Sum("total_count")) |
| 144 | ) |
| 145 | result: dict[int, dict[str, int]] = defaultdict(lambda: defaultdict(int)) |
| 146 | for row in qs: |
| 147 | resource = Resource(row["resource"]) |
| 148 | result[row["environment_id"]][resource.resource_name] = row["total"] |
| 149 | return result |
| 150 | |
| 151 | |
| 152 | def _get_env_ids_for_orgs( |
no test coverage detected
searching dependent graphs…