Function get_organisation_metrics

api/platform_hub/services.py:165–363

(
    organisations: QuerySet[Organisation],
)

def get_organisation_metrics(
    organisations: QuerySet[Organisation],
) -> list[OrganisationMetricsData]:
    cutoff_30d = timezone.now() - timedelta(days=30)
    org_ids = set(organisations.values_list("id", flat=True))

    orgs = organisations.prefetch_related(
        "projects__environments",
        "users",
    ).order_by("id")

    env_ids_by_org = _get_env_ids_for_orgs(organisations)
    all_env_ids = [eid for eids in env_ids_by_org.values() for eid in eids]

    # API usage data per org (30d/60d/90d)
    org_usage_30d: dict[int, int] = {}
    org_usage_60d: dict[int, int] = {}
    org_usage_90d: dict[int, int] = {}
    env_usage: dict[int, dict[str, int]] = {}
    env_usage_30d: dict[int, dict[str, int]] = {}

    if settings.USE_POSTGRES_FOR_ANALYTICS:
        env_usage = _get_api_usage_for_envs_postgres(all_env_ids, days=90)
        env_usage_30d = _get_api_usage_for_envs_postgres(all_env_ids, days=30)
        env_usage_60d = _get_api_usage_for_envs_postgres(all_env_ids, days=60)

        for oid, eids in env_ids_by_org.items():
            org_usage_30d[oid] = sum(
                sum(env_usage_30d.get(eid, {}).values()) for eid in eids
            )
            org_usage_60d[oid] = sum(
                sum(env_usage_60d.get(eid, {}).values()) for eid in eids
            )
            org_usage_90d[oid] = sum(
                sum(env_usage.get(eid, {}).values()) for eid in eids
            )
    elif settings.INFLUXDB_TOKEN:
        org_usage_30d = _get_api_usage_for_orgs_influx(org_ids, days=30)
        org_usage_60d = _get_api_usage_for_orgs_influx(org_ids, days=60)
        org_usage_90d = _get_api_usage_for_orgs_influx(org_ids, days=90)
    else:
        logger.warning("no-analytics-database-configured")

    # Allowed API calls per org
    sub_caches = {
        sc.organisation_id: sc
        for sc in OrganisationSubscriptionInformationCache.objects.filter(
            organisation__in=organisations,
        )
    }

    # Flag counts per org
    flag_counts = dict(
        Feature.objects.filter(project__organisation__in=organisations)
        .values("project__organisation_id")
        .annotate(count=Count("id"))
        .values_list("project__organisation_id", "count")
    )