I've been working on a task to apply throttling on a flask app. For throttling, I've been looking into flask-limiter. My app has all endpoints extended from flask-restful's Resource.

class CompanyApi(Resource):
    decorators = [limiter.limit(limit_value="10/minute")]

    def get(self):
        return "successful"

From flask-limiter doc, its clearly mentioned that dynamic limits can be loaded using callable in decorator for method based views.

def get_limit():
   company = request.args.get('company')
   limit = Company.query.get(company)
   #returns a limit string stored in db
   return limit

@app.route("/check_company", methods=["GET"])
def check_company():
    return "success"

While for pluggable views, its only provided the hardcoded example by setting the decorator as:

decorators = [limiter.limit(limit_value="10/minute")]

I've tried by setting default value in decorators and when request is processing, I get request params (company) on which limit is retrieved from db. Then overwriting limiter's limit value:

CompanyApi.decorators = [Limiter.limit(limit_value=get_limit)]

It's been changed but not effective. I need to set limit for each endpoint dynamically on the basis of request.

How can I achieve to set dynamic limits for class based views?

1 Answers

Azeem On Best Solutions

I've been looking into flask-limiter issues and found someone has tried custom limiter. I had found a workaround to meet my requirements. Custom limiter has following code:

class CustomLimiter(Limiter):
    def __init__(self):
            key_func=lambda: str(g.user.id) if hasattr(g, 'user') else get_ipaddr(),

    def _evaluate_limits(self, limits):
        failed_limit = None
        limit_for_header = None
        for lim in limits:
            limit_scope = request.endpoint
            limit_key = lim.key_func()
            assert limit_key, 'key expected'
            args = [limit_key, limit_scope]

            if self._key_prefix:
                args = [self._key_prefix] + args
            if not limit_for_header or lim.limit < limit_for_header[0]:
                limit_for_header = [lim.limit] + args
            if not self.limiter.hit(lim.limit, *args):
                    "ratelimit %s (%s) exceeded at endpoint: %s",
                    lim.limit, limit_key, limit_scope
                failed_limit = lim
                limit_for_header = [lim.limit] + args
        g.view_rate_limit = limit_for_header

        if failed_limit:
            raise RateLimitExceeded(failed_limit.limit)

    def limit(self, limit_value, key_func=None):
        def _inner(obj):
            assert not isinstance(obj, Blueprint)
            func = key_func or self._key_func

            if callable(limit_value):
                limits = [LimitGroup(limit_value, func, None, False, None, None, None)]
                limits = list(LimitGroup(limit_value, func, None, False, None, None, None))

            def __inner(*a, **k):
                return obj(*a, **k)
            return __inner
        return _inner

limiter = CustomLimiter()

I added a check into _evaluate_limits:

def _evaluate_limits(self, limits):
        if request:
            company = request.args.get('company')
            limit = Company.query.get(company)
            limits = list(LimitGroup(
                get_company_name,  # a callable as a key_func

In this modification, limiter always sets defaults limits to process whenever the Api is instantiated, but whenever there is a request, it will check and replace the limits by creating a key using key function. Key insures the counter for next time for throttling.

In this way I was able to achieve dynamic limit behaviour for pluggable views.