pyrate_limiter.buckets.redis_bucket module

Bucket implementation using Redis

class pyrate_limiter.buckets.redis_bucket.LuaScript

Bases: object

Scripts that deal with bucket operations

PUT_ITEM = '\n    local bucket = KEYS[1]\n    local now = ARGV[1]\n    local space_required = tonumber(ARGV[2])\n    local item_name = ARGV[3]\n    local rates_count = tonumber(ARGV[4])\n\n    -- Per rate the client supplies (window_start, limit, blocking_rank); the\n    -- window bounds and rank come from the algorithm, so this script stays\n    -- policy-agnostic. blocking_rank < 0 means "resolve no entry".\n    for i=1,rates_count do\n        local offset = (i - 1) * 3\n        local window_start = tonumber(ARGV[5 + offset])\n        local limit = tonumber(ARGV[5 + offset + 1])\n        local blocking_rank = tonumber(ARGV[5 + offset + 2])\n        local count = redis.call(\'ZCOUNT\', bucket, window_start, now)\n        local space_available = limit - tonumber(count)\n        if space_available < space_required then\n            local blocking_timestamp = -1\n\n            if blocking_rank >= 0 then\n                local blocking = redis.call(\'ZRANGE\', bucket, -1 - blocking_rank, -1 - blocking_rank, \'WITHSCORES\')\n                if blocking[2] then\n                    blocking_timestamp = tonumber(blocking[2])\n                end\n            end\n\n            return {i - 1, blocking_timestamp}\n        end\n    end\n\n    local batch = {}\n    -- Each member adds two unpacked arguments; 1000 stays below Lua 5.1 limits.\n    local batch_size = 1000\n\n    for i=1,space_required do\n        batch[#batch + 1] = now\n        batch[#batch + 1] = item_name..i\n\n        if #batch == batch_size * 2 then\n            redis.call(\'ZADD\', bucket, unpack(batch))\n            batch = {}\n        end\n    end\n\n    if #batch > 0 then\n        redis.call(\'ZADD\', bucket, unpack(batch))\n    end\n\n    return {-1, -1}\n    '

class pyrate_limiter.buckets.redis_bucket.RedisBucket(rates, redis, bucket_key, script_hash, algorithm=None)

Bases: AbstractBucket

A bucket using redis for storing data - We are not using redis’ built-in TIME since it is non-deterministic - In distributed context, use local server time or a remote time server - Each bucket instance use a dedicated connection to avoid race-condition - can be either sync or async

bucket_key

count()

Count number of items in the bucket

flush()

Flush the whole bucket - Must remove failing-rate after flushing

classmethod init(rates, redis, bucket_key, algorithm=None)

leak(current_timestamp=None)

leaking bucket - removing items that are outdated

Return type:

int | Awaitable[int]

now()

Retrieve current timestamp from the clock backend.

peek(index)

Peek at the rate-item at a specific index in latest-to-earliest order NOTE: The reason we cannot peek from the start of the queue(earliest-to-latest) is we can’t really tell how many outdated items are still in the queue

Return type:

RateItem | None | Awaitable[RateItem | None]

put(item)

Add item to key

Return type:

bool | Awaitable[bool]

redis

script_hash