feat: Create api usage function (#3340)

api/app_analytics/influxdb_wrapper.py

@@ -88,7 +88,6 @@ def influx_query_manager(
             f" |> drop(columns: {drop_columns_input})"
             f"{extra}"
         )
-
         logger.debug("Running query in influx: \n\n %s", query)
 
         try:
@@ -314,6 +313,46 @@ def get_top_organisations(date_range: str, limit: str = ""):
     return dataset
 
 
+def get_current_api_usage(organisation_id: int, date_range: str) -> int:
+    """
+    Query influx db for api usage
+
+    :param organisation_id: filtered organisation
+    :param date_range: data range for current api usage window
+
+    :return: number of current api calls
+    """
+
+    bucket = read_bucket
+    results = InfluxDBWrapper.influx_query_manager(
+        date_range=date_range,
+        bucket=bucket,
+        filters=build_filter_string(
+            [
+                'r._measurement == "api_call"',
+                'r["_field"] == "request_count"',
+                f'r["organisation_id"] == "{organisation_id}"',
+            ]
+        ),
+        drop_columns=("_start", "_stop", "_time"),
+        extra='|> sum() \
+              |> group() \
+              |> sort(columns: ["_value"], desc: true) ',
+    )
+
+    for result in results:
+        # Return zero if there are no API calls recorded.
+        if len(result.records) == 0:
+            return 0
+
+        # There should only be one matching result due to the
+        # group part of the query.
+        assert len(result.records) == 1
+        return result.records[0].get_value()
+
+    return 0
+
+
 def build_filter_string(filter_expressions: typing.List[str]) -> str:
     return "|> ".join(
         ["", *[f"filter(fn: (r) => {exp})" for exp in filter_expressions]]

api/organisations/chargebee/serializers.py

@@ -33,6 +33,7 @@ class ProcessSubscriptionSubscriptionSerializer(Serializer):
     id = CharField(allow_null=False)
     status = CharField(allow_null=False)
     plan_id = CharField(allow_null=True, required=False, default=None)
+    current_term_start = IntegerField(required=False, default=None)
     current_term_end = IntegerField(required=False, default=None)
     addons = ListField(
         child=ProcessSubscriptionAddonsSerializer(), required=False, default=list

api/organisations/chargebee/webhook_handlers.py

@@ -103,7 +103,7 @@ def payment_succeeded(request: Request) -> Response:
     return Response(status=status.HTTP_200_OK)
 
 
-def process_subscription(request: Request) -> Response:
+def process_subscription(request: Request) -> Response:  # noqa: C901
     serializer = ProcessSubscriptionSerializer(data=request.data)
 
     # Since this function is a catchall, we're not surprised if
@@ -145,15 +145,35 @@ def process_subscription(request: Request) -> Response:
         chargebee_subscription=subscription,
         customer_email=customer["email"],
     )
+    osic_defaults = {
+        "chargebee_updated_at": timezone.now(),
+        "allowed_30d_api_calls": subscription_metadata.api_calls,
+        "allowed_seats": subscription_metadata.seats,
+        "organisation_id": existing_subscription.organisation_id,
+        "allowed_projects": subscription_metadata.projects,
+        "chargebee_email": subscription_metadata.chargebee_email,
+    }
+
+    if "current_term_end" in subscription:
+        current_term_end = subscription["current_term_end"]
+        if current_term_end is None:
+            osic_defaults["current_billing_term_ends_at"] = None
+        else:
+            osic_defaults["current_billing_term_ends_at"] = datetime.fromtimestamp(
+                current_term_end
+            ).replace(tzinfo=timezone.utc)
+
+    if "current_term_start" in subscription:
+        current_term_start = subscription["current_term_start"]
+        if current_term_start is None:
+            osic_defaults["current_billing_term_starts_at"] = None
+        else:
+            osic_defaults["current_billing_term_starts_at"] = datetime.fromtimestamp(
+                current_term_start
+            ).replace(tzinfo=timezone.utc)
+
     OrganisationSubscriptionInformationCache.objects.update_or_create(
         organisation_id=existing_subscription.organisation_id,
-        defaults={
-            "chargebee_updated_at": timezone.now(),
-            "allowed_30d_api_calls": subscription_metadata.api_calls,
-            "allowed_seats": subscription_metadata.seats,
-            "organisation_id": existing_subscription.organisation_id,
-            "allowed_projects": subscription_metadata.projects,
-            "chargebee_email": subscription_metadata.chargebee_email,
-        },
+        defaults=osic_defaults,
     )
     return Response(status=status.HTTP_200_OK)

api/organisations/constants.py

@@ -0,0 +1,5 @@
+API_USAGE_ALERT_THRESHOLDS = [75, 90, 100, 120]
+ALERT_EMAIL_MESSAGE = (
+    "Organisation %s has used %d seats which is over their plan limit of %d (plan: %s)"
+)
+ALERT_EMAIL_SUBJECT = "Organisation over number of seats"

api/organisations/migrations/0051_create_org_api_usage_notification.py

@@ -0,0 +1,36 @@
+# Generated by Django 3.2.23 on 2024-02-05 16:53
+
+import django.core.validators
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('organisations', '0050_add_historical_subscription'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='organisationsubscriptioninformationcache',
+            name='current_billing_term_ends_at',
+            field=models.DateTimeField(null=True),
+        ),
+        migrations.AddField(
+            model_name='organisationsubscriptioninformationcache',
+            name='current_billing_term_starts_at',
+            field=models.DateTimeField(null=True),
+        ),
+        migrations.CreateModel(
+            name='OranisationAPIUsageNotification',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('percent_usage', models.IntegerField(validators=[django.core.validators.MinValueValidator(75), django.core.validators.MaxValueValidator(120)])),
+                ('notified_at', models.DateTimeField(null=True)),
+                ('created_at', models.DateTimeField(auto_now_add=True, null=True)),
+                ('updated_at', models.DateTimeField(auto_now=True, null=True)),
+                ('organisation', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='api_usage_notifications', to='organisations.organisation')),
+            ],
+        ),
+    ]

api/organisations/models.py

@@ -4,6 +4,7 @@
 from core.models import SoftDeleteExportableModel
 from django.conf import settings
 from django.core.cache import caches
+from django.core.validators import MaxValueValidator, MinValueValidator
 from django.db import models
 from django.utils import timezone
 from django_lifecycle import (
@@ -415,6 +416,8 @@ class OrganisationSubscriptionInformationCache(models.Model):
     updated_at = models.DateTimeField(auto_now=True)
     chargebee_updated_at = models.DateTimeField(auto_now=False, null=True)
     influx_updated_at = models.DateTimeField(auto_now=False, null=True)
+    current_billing_term_starts_at = models.DateTimeField(auto_now=False, null=True)
+    current_billing_term_ends_at = models.DateTimeField(auto_now=False, null=True)
 
     api_calls_24h = models.IntegerField(default=0)
     api_calls_7d = models.IntegerField(default=0)
@@ -425,3 +428,17 @@ class OrganisationSubscriptionInformationCache(models.Model):
     allowed_projects = models.IntegerField(default=1, blank=True, null=True)
 
     chargebee_email = models.EmailField(blank=True, max_length=254, null=True)
+
+
+class OranisationAPIUsageNotification(models.Model):
+    organisation = models.ForeignKey(
+        Organisation, on_delete=models.CASCADE, related_name="api_usage_notifications"
+    )
+    percent_usage = models.IntegerField(
+        null=False,
+        validators=[MinValueValidator(75), MaxValueValidator(120)],
+    )
+    notified_at = models.DateTimeField(null=True)
+
+    created_at = models.DateTimeField(null=True, auto_now_add=True)
+    updated_at = models.DateTimeField(null=True, auto_now=True)

api/organisations/tasks.py

@@ -1,9 +1,20 @@
+import logging
 from datetime import timedelta
 
+from app_analytics.influxdb_wrapper import get_current_api_usage
+from dateutil.relativedelta import relativedelta
+from django.conf import settings
+from django.core.mail import send_mail
+from django.template.loader import render_to_string
 from django.utils import timezone
 
 from organisations import subscription_info_cache
-from organisations.models import Organisation, Subscription
+from organisations.models import (
+    OranisationAPIUsageNotification,
+    Organisation,
+    OrganisationRole,
+    Subscription,
+)
 from organisations.subscriptions.subscription_service import (
     get_subscription_metadata,
 )
@@ -13,12 +24,14 @@
 )
 from users.models import FFAdminUser
 
+from .constants import (
+    ALERT_EMAIL_MESSAGE,
+    ALERT_EMAIL_SUBJECT,
+    API_USAGE_ALERT_THRESHOLDS,
+)
 from .subscriptions.constants import SubscriptionCacheEntity
 
-ALERT_EMAIL_MESSAGE = (
-    "Organisation %s has used %d seats which is over their plan limit of %d (plan: %s)"
-)
-ALERT_EMAIL_SUBJECT = "Organisation over number of seats"
+logger = logging.getLogger(__name__)
 
 
 @register_task_handler()
@@ -73,3 +86,97 @@ def finish_subscription_cancellation():
     ):
         subscription.organisation.cancel_users()
         subscription.save_as_free_subscription()
+
+
+def send_admin_api_usage_notification(
+    organisation: Organisation, matched_threshold: int
+) -> None:
+    """
+    Send notification to admins that the API has breached a threshold.
+    """
+
+    recipient_list = FFAdminUser.objects.filter(
+        userorganisation__organisation=organisation,
+    )
+
+    if matched_threshold < 100:
+        message = "organisations/api_usage_notification.txt"
+        html_message = "organisations/api_usage_notification.html"
+
+        # Since threshold < 100 only include admins.
+        recipient_list = recipient_list.filter(
+            userorganisation__role=OrganisationRole.ADMIN,
+        )
+    else:
+        message = "organisations/api_usage_notification_limit.txt"
+        html_message = "organisations/api_usage_notification_limit.html"
+
+    context = {
+        "organisation": organisation,
+        "matched_threshold": matched_threshold,
+    }
+
+    send_mail(
+        subject=f"Flagsmith API use has reached {matched_threshold}%",
+        message=render_to_string(message, context),
+        from_email=settings.DEFAULT_FROM_EMAIL,
+        recipient_list=list(recipient_list.values_list("email", flat=True)),
+        html_message=render_to_string(html_message, context),
+        fail_silently=True,
+    )
+
+    OranisationAPIUsageNotification.objects.create(
+        organisation=organisation,
+        percent_usage=matched_threshold,
+        notified_at=timezone.now(),
+    )
+
+
+def _handle_api_usage_notifications(organisation: Organisation):
+    subscription_cache = organisation.subscription_information_cache
+    billing_starts_at = subscription_cache.current_billing_term_starts_at
+    now = timezone.now()
+
+    # Truncate to the closest active month to get start of current period.
+    month_delta = relativedelta(now, billing_starts_at).months
+    period_starts_at = relativedelta(months=month_delta) + billing_starts_at
+
+    days = relativedelta(now, period_starts_at).days
+    api_usage = get_current_api_usage(organisation.id, f"{days}d")
+
+    api_usage_percent = int(100 * api_usage / subscription_cache.allowed_30d_api_calls)
+
+    matched_threshold = None
+    for threshold in API_USAGE_ALERT_THRESHOLDS:
+        if threshold > api_usage_percent:
+            break
+
+        matched_threshold = threshold
+
+    if OranisationAPIUsageNotification.objects.filter(
+        notified_at__gt=period_starts_at,
+        percent_usage=matched_threshold,
+    ).exists():
+        # Already sent the max notification level so don't resend.
+        return
+
+    send_admin_api_usage_notification(organisation, matched_threshold)
+
+
+@register_recurring_task(
+    run_every=timedelta(hours=12),
+)
+def handle_api_usage_notifications():
+    for organisation in Organisation.objects.filter(
+        subscription_information_cache__current_billing_term_starts_at__isnull=False,
+        subscription_information_cache__current_billing_term_ends_at__isnull=False,
+    ).select_related(
+        "subscription_information_cache",
+    ):
+        try:
+            _handle_api_usage_notifications(organisation)
+        except RuntimeError:
+            logger.error(
+                f"Error processing api usage for organisation {organisation.id}",
+                exc_info=True,
+            )

api/organisations/templates/organisations/api_usage_notification.html

@@ -0,0 +1,32 @@
+<table>
+
+        <tr>
+
+               <td>Hi there,</td>
+
+        </tr>
+
+        <tr>
+
+               <td>
+                 The API usage for {{ organisation.name }} has reached
+                 {{ matched_threshold }}% within the current subscription period.
+                 Please consider upgrading your organisations account limits.
+               </td>
+
+
+        </tr>
+
+        <tr>
+
+               <td>Thank you!</td>
+
+        </tr>
+
+        <tr>
+
+               <td>The Flagsmith Team</td>
+
+        </tr>
+
+</table>

api/organisations/templates/organisations/api_usage_notification.txt

@@ -0,0 +1,7 @@
+Hi there,
+
+The API usage for {{ organisation.name }} has reached {{ matched_threshold }}% within the current subscription period. Please consider upgrading your organisations account limits.
+
+Thank you!
+
+The Flagsmith Team