api.rate_limit

Complexity

Total Complexity

12

Size/Duplication

Total Lines	60
Duplicated Lines	0 %

Importance

Changes

0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	FakeRedis.incr()	0	5	2
A	FakeRedis.__init__()	0	3	1
A	FakeRedis.pipeline()	0	2	1
A	FakeRedis.expireat()	0	2	1
A	FakeRedis.execute()	0	2	1
A	RateLimit.over_limit()	0	3	1
A	RateLimit.remaining()	0	3	1
A	RateLimit.__init__()	0	19	4

import time
from redis import Redis
from flask import current_app

redis = None


class FakeRedis:
  """Redis mock used for testing."""
  def __init__(self):
    self.v = {}
    self.last_key = None

  def pipeline(self):
    return self

  def incr(self, key):
    if self.v.get(key, None) is None:
      self.v[key] = 0
    self.v[key] += 1
    self.last_key = key

  def expireat(self, key, exp_time):
    pass

  def execute(self):
    return [self.v[self.last_key]]


class RateLimit:
  expiration_window = 10

  def __init__(self, key_prefix, limit, per):
    global redis
    if redis is None and current_app.config['USE_RATE_LIMITS']:
      if current_app.config['TESTING']:
        redis = FakeRedis()
      else:  # pragma: no cover
        redis_host = current_app.config.get("REDIS_HOST", "localhost")
        redis_port = current_app.config.get("REDIS_PORT", 6379)
        redis_db = current_app.config.get("REDIS_DB", 0)
        redis = Redis(host=redis_host, port=redis_port, db=redis_db)

    self.reset = (int(time.time()) // per) * per + per
    self.key = key_prefix + str(self.reset)
    self.limit = limit
    self.per = per
    p = redis.pipeline()
    p.incr(self.key)
    p.expireat(self.key, self.reset + self.expiration_window)
    self.current = min(p.execute()[0], limit)

  @property
  def remaining(self):
    return self.limit - self.current

  @property
  def over_limit(self):
    return self.current >= self.limit