Function handle_api_usage_notifications

api/organisations/tasks.py:121–155

()

# Task enqueued in register_recurring_tasks below.
def handle_api_usage_notifications() -> None:
    openfeature_client = get_openfeature_client()

    threshold_percentage = min(API_USAGE_ALERT_THRESHOLDS) / 100
    for organisation in Organisation.objects.filter(
        # Not having a subscription information cache implies that the organisation has
        # no usage so no notification is needed.
        subscription_information_cache__isnull=False,
        # Skip organisations whose usage in the last 30d is below the threshold of the
        # minimum notification level for their allowance. If their usage is below the
        # threshold in the last 30d, it must be below it for the current billing period
        # (which by definition is <30d).
        subscription_information_cache__api_calls_30d__gte=F(
            "subscription_information_cache__allowed_30d_api_calls"
        )
        * threshold_percentage,
    ).select_related("subscription", "subscription_information_cache"):
        feature_enabled = openfeature_client.get_boolean_value(
            "api_usage_alerting",
            default_value=False,
            evaluation_context=organisation.openfeature_evaluation_context,
        )
        if not feature_enabled:
            logger.info(
                f"Skipping processing API usage for organisation {organisation.id}"
            )
            continue

        try:
            handle_api_usage_notification_for_organisation(organisation)
        except Exception:
            logger.error(
                f"Error processing api usage for organisation {organisation.id}",
                exc_info=True,
            )


# Task enqueued in register_recurring_tasks below.