Skip to content

Storage API

Storage Backends

Memory Backends

aiogram_sentinel.storage.MemoryRateLimiter 

MemoryRateLimiter()

Bases: RateLimiterBackend

In-memory rate limiter using sliding window with TTL cleanup.

Initialize the rate limiter.

Source code in src/aiogram_sentinel/storage/memory.py 
15
16
17
18
def __init__(self) -> None:
    """Initialize the rate limiter."""
    self._counters: dict[str, deque[float]] = defaultdict(deque)
    self._lock = asyncio.Lock()

allow async 

allow(key, max_events, per_seconds)

Check if request is allowed and increment counter.

Source code in src/aiogram_sentinel/storage/memory.py 
20
21
22
23
24
25
26
27
28
29
30
31
async def allow(self, key: str, max_events: int, per_seconds: int) -> bool:
    """Check if request is allowed and increment counter."""
    async with self._lock:
        now = time.monotonic()
        # Clean up old entries
        self._cleanup_old_entries(key, now, per_seconds)
        # Check if under limit
        if len(self._counters[key]) < max_events:
            # Add current timestamp
            self._counters[key].append(now)
            return True
        return False

get_rate_limit async 

get_rate_limit(key)

Get current rate limit count for key.

Source code in src/aiogram_sentinel/storage/memory.py 
61
62
63
64
65
66
67
async def get_rate_limit(self, key: str) -> int:
    """Get current rate limit count for key."""
    async with self._lock:
        now = time.monotonic()
        # Clean up old entries (use a reasonable default window)
        self._cleanup_old_entries(key, now, 60)  # 60 second default window
        return len(self._counters[key])

get_remaining async 

get_remaining(key, max_events, per_seconds)

Get remaining requests in current window.

Source code in src/aiogram_sentinel/storage/memory.py 
33
34
35
36
37
38
39
40
async def get_remaining(self, key: str, max_events: int, per_seconds: int) -> int:
    """Get remaining requests in current window."""
    async with self._lock:
        now = time.monotonic()
        # Clean up old entries
        self._cleanup_old_entries(key, now, per_seconds)
        current_count = len(self._counters[key])
        return max(0, max_events - current_count)

increment_rate_limit async 

increment_rate_limit(key, window)

Increment rate limit counter and return current count.

Source code in src/aiogram_sentinel/storage/memory.py 
51
52
53
54
55
56
57
58
59
async def increment_rate_limit(self, key: str, window: int) -> int:
    """Increment rate limit counter and return current count."""
    async with self._lock:
        now = time.monotonic()
        # Clean up old entries
        self._cleanup_old_entries(key, now, window)
        # Add current timestamp
        self._counters[key].append(now)
        return len(self._counters[key])

reset_rate_limit async 

reset_rate_limit(key)

Reset rate limit for key.

Source code in src/aiogram_sentinel/storage/memory.py 
69
70
71
72
73
async def reset_rate_limit(self, key: str) -> None:
    """Reset rate limit for key."""
    async with self._lock:
        if key in self._counters:
            self._counters[key].clear()

aiogram_sentinel.storage.MemoryDebounce 

MemoryDebounce()

Bases: DebounceBackend

In-memory debounce backend using monotonic time.

Initialize the debounce backend.

Source code in src/aiogram_sentinel/storage/memory.py 
79
80
81
82
def __init__(self) -> None:
    """Initialize the debounce backend."""
    self._store: dict[str, float] = {}
    self._lock = asyncio.Lock()

is_debounced async 

is_debounced(key)

Check if key is currently debounced.

Source code in src/aiogram_sentinel/storage/memory.py 
107
108
109
110
111
112
113
114
115
116
117
async def is_debounced(self, key: str) -> bool:
    """Check if key is currently debounced."""
    async with self._lock:
        now = time.monotonic()
        ts = self._store.get(key, 0)
        if ts and ts >= now:  # Use >= for boundary case
            return True
        # Clean up expired entries
        if key in self._store:
            del self._store[key]
        return False

seen async 

seen(key, window_seconds, fingerprint)

Check if fingerprint was seen within window and record it.

Source code in src/aiogram_sentinel/storage/memory.py 
84
85
86
87
88
89
90
91
92
93
async def seen(self, key: str, window_seconds: int, fingerprint: str) -> bool:
    """Check if fingerprint was seen within window and record it."""
    k = f"{key}:{fingerprint}"
    async with self._lock:
        now = time.monotonic()
        ts = self._store.get(k, 0)
        if ts and ts + window_seconds > now:
            return True
        self._store[k] = now
        return False

set_debounce async 

set_debounce(key, delay)

Set debounce for a key.

Source code in src/aiogram_sentinel/storage/memory.py 
 96
 97
 98
 99
100
101
102
103
104
105
async def set_debounce(self, key: str, delay: float) -> None:
    """Set debounce for a key."""
    async with self._lock:
        now = time.monotonic()
        if delay <= 0:
            # For zero or negative delay, don't set debounce
            if key in self._store:
                del self._store[key]
        else:
            self._store[key] = now + delay

Redis Backends

aiogram_sentinel.storage.RedisRateLimiter 

RedisRateLimiter(redis, prefix)

Bases: RateLimiterBackend

Redis rate limiter using INCR + EXPIRE pattern.

Initialize the rate limiter.

Source code in src/aiogram_sentinel/storage/redis.py 
22
23
24
25
def __init__(self, redis: Redis, prefix: str) -> None:
    """Initialize the rate limiter."""
    self._redis = redis
    self._prefix = prefix

allow async 

allow(key, max_events, per_seconds)

Check if request is allowed and increment counter.

Source code in src/aiogram_sentinel/storage/redis.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
async def allow(self, key: str, max_events: int, per_seconds: int) -> bool:
    """Check if request is allowed and increment counter."""
    try:
        redis_key = _k(self._prefix, "rate", key)
        # Use pipeline for atomic operation
        pipe = self._redis.pipeline()
        pipe.incr(redis_key)
        pipe.ttl(redis_key)
        count, ttl = await pipe.execute()
        if ttl == -1:  # Set TTL if absent
            await self._redis.expire(redis_key, per_seconds)
        return int(count) <= max_events
    except RedisError as e:
        raise BackendOperationError(f"Failed to check rate limit: {e}") from e

get_remaining async 

get_remaining(key, max_events, per_seconds)

Get remaining requests in current window.

Source code in src/aiogram_sentinel/storage/redis.py 
42
43
44
45
46
47
48
49
async def get_remaining(self, key: str, max_events: int, per_seconds: int) -> int:
    """Get remaining requests in current window."""
    try:
        redis_key = _k(self._prefix, "rate", key)
        val = await self._redis.get(redis_key)
        return max(0, max_events - int(val or 0))
    except RedisError as e:
        raise BackendOperationError(f"Failed to get remaining: {e}") from e

aiogram_sentinel.storage.RedisDebounce 

RedisDebounce(redis, prefix)

Bases: DebounceBackend

Redis debounce backend using SET NX EX pattern.

Initialize the debounce backend.

Source code in src/aiogram_sentinel/storage/redis.py 
55
56
57
58
def __init__(self, redis: Redis, prefix: str) -> None:
    """Initialize the debounce backend."""
    self._redis = redis
    self._prefix = prefix

seen async 

seen(key, window_seconds, fingerprint)

Check if fingerprint was seen within window and record it.

Source code in src/aiogram_sentinel/storage/redis.py 
60
61
62
63
64
65
66
67
68
69
70
71
async def seen(self, key: str, window_seconds: int, fingerprint: str) -> bool:
    """Check if fingerprint was seen within window and record it."""
    try:
        fp = fingerprint  # Use fingerprint directly
        k = _k(self._prefix, "debounce", key, fp)
        added = await self._redis.set(
            k, int(time.time()), ex=window_seconds, nx=True
        )
        # nx=True => returns True if set, None if exists
        return added is None
    except RedisError as e:
        raise BackendOperationError(f"Failed to check debounce: {e}") from e

Storage Protocols

aiogram_sentinel.storage.base.RateLimiterBackend

Bases: Protocol

Protocol for rate limiting storage backend.

allow async 

allow(key, max_events, per_seconds)

Check if request is allowed and increment counter.

Source code in src/aiogram_sentinel/storage/base.py 
12
13
14
async def allow(self, key: str, max_events: int, per_seconds: int) -> bool:
    """Check if request is allowed and increment counter."""
    ...

get_remaining async 

get_remaining(key, max_events, per_seconds)

Get remaining requests in current window.

Source code in src/aiogram_sentinel/storage/base.py 
16
17
18
async def get_remaining(self, key: str, max_events: int, per_seconds: int) -> int:
    """Get remaining requests in current window."""
    ...

aiogram_sentinel.storage.base.DebounceBackend

Bases: Protocol

Protocol for debouncing storage backend.

seen async 

seen(key, window_seconds, fingerprint)

Check if fingerprint was seen within window and record it.

Source code in src/aiogram_sentinel/storage/base.py 
25
26
27
async def seen(self, key: str, window_seconds: int, fingerprint: str) -> bool:
    """Check if fingerprint was seen within window and record it."""
    ...

Factory

aiogram_sentinel.storage.factory.build_infra 

build_infra(config)

Build infrastructure backends (rate_limiter + debounce) based on configuration.

Source code in src/aiogram_sentinel/storage/factory.py 
10
11
12
13
14
15
16
17
def build_infra(config: SentinelConfig) -> InfraBundle:
    """Build infrastructure backends (rate_limiter + debounce) based on configuration."""
    if config.backend == "memory":
        return _build_memory_infra()
    elif config.backend == "redis":
        return _build_redis_infra(config)
    else:
        raise ConfigurationError(f"Unsupported backend: {config.backend}")