Function get_current_api_usage

api/app_analytics/influxdb_wrapper.py:439–475

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

(
    organisation_id: int,
    date_start: datetime,
)

def get_current_api_usage(
    organisation_id: int,
    date_start: datetime,
) -> 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 = InfluxDBWrapper.get_downsampled_bucket(DownsampleSize.FIFTEEN_MINUTES)
    results = InfluxDBWrapper.influx_query_manager(
        date_start=date_start,
        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

        return sum(r.get_value() for r in result.records)

    return 0


def get_platform_usage_trends(