Decorator to measure call rate

rate_measured.py

# -*- coding: utf-8 -*-

import random
import json
import sys
import time


def rate_measured(report_period=10, report_function=None):
    """
    Function decorator which measures the call rate of the decorated function.

    :param report_period: Number of seconds between each report
    :param report_function: Function to pass in a JSON blob as sole arg
    """
    def decorate(func):
        start_time = int(time.time())
        counter_time = start_time
        counters = []

        for td in range(report_period + 1):
            counters.append(0)

        def _report_metrics(now):
            nonlocal counter_time, counters, report_period

            max_rate = 0
            min_rate = sys.maxsize
            sum_calls = 0

            for t in range(report_period):
                sum_calls += counters[t]
                max_rate = max(max_rate, counters[t])
                min_rate = min(min_rate, counters[t])
                counters[t] = 0

            counter_time = now

            # Report findings
            data = json.dumps({
                'start_timestamp': counter_time,
                'end_timestamp': counter_time + report_period,
                'report_period': report_period,
                'sum_calls': sum_calls,
                'avg_rate': 1.0 * sum_calls / report_period,
                'max_rate': max_rate,
                'min_rate': min_rate,
            })
            if report_function:
                report_function(data)
            else:
                print(data)

        def rate_measured_function(*args, **kwargs):
            nonlocal start_time, counter_time, counters, report_period

            timestamp = int(time.time())

            if timestamp >= counter_time + report_period:
                _report_metrics(timestamp)

            delta = (timestamp - start_time) % report_period
            try:
                counters[delta] += 1
            except IndexError:
                counters[delta] = 1

            return func(*args, **kwargs)

        return rate_measured_function

    return decorate


if __name__ == '__main__':

    @rate_measured()
    def my_func(a, b):
        time.sleep(random.random() / 500)


    for i in range(100000):
        my_func(1, 20)