Skip to content

Core API

Error Handling

aiogram_sentinel.ErrorConfig dataclass 

ErrorConfig(use_friendly_messages=True, domain_classifier=None, message_resolver=None, locale_resolver=None, on_error=None, sync_retry_after=True, respond_strategy='answer', show_alert_for_callbacks=False, send_strategy=None)

Configuration for error handling middleware.

use_friendly_messages class-attribute instance-attribute 

use_friendly_messages = True

domain_classifier class-attribute instance-attribute 

domain_classifier = None

message_resolver class-attribute instance-attribute 

message_resolver = None

locale_resolver class-attribute instance-attribute 

locale_resolver = None

on_error class-attribute instance-attribute 

on_error = None

sync_retry_after class-attribute instance-attribute 

sync_retry_after = True

respond_strategy class-attribute instance-attribute 

respond_strategy = 'answer'

show_alert_for_callbacks class-attribute instance-attribute 

show_alert_for_callbacks = False

send_strategy class-attribute instance-attribute 

send_strategy = None

aiogram_sentinel.ErrorEvent dataclass 

ErrorEvent(error_type, error_message, event_type, user_id, chat_id, locale, retry_after=None)

Bases: BaseEvent

Event emitted when an error occurs.

error_type instance-attribute 

error_type

error_message instance-attribute 

error_message

event_type instance-attribute 

event_type

user_id instance-attribute 

user_id

chat_id instance-attribute 

chat_id

locale instance-attribute 

locale

retry_after class-attribute instance-attribute 

retry_after = None

aiogram_sentinel.ErrorHandlingMiddleware 

ErrorHandlingMiddleware(cfg, key_builder, rate_limiter=None)

Bases: BaseMiddleware

Middleware for centralized error handling.

Initialize the error handling middleware.

Parameters:

Name Type Description Default
cfg ErrorConfig

Error configuration

 required
key_builder KeyBuilder

KeyBuilder instance for key generation

 required
rate_limiter RateLimiterBackend | None

Optional rate limiter for RetryAfter sync

 None
Source code in src/aiogram_sentinel/middlewares/errors.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def __init__(
    self,
    cfg: ErrorConfig,
    key_builder: KeyBuilder,
    rate_limiter: RateLimiterBackend | None = None,
) -> None:
    """Initialize the error handling middleware.

    Args:
        cfg: Error configuration
        key_builder: KeyBuilder instance for key generation
        rate_limiter: Optional rate limiter for RetryAfter sync
    """
    super().__init__()
    self._cfg = cfg
    self._key_builder = key_builder
    self._rate_limiter = rate_limiter

__call__ async 

__call__(handler, event, data)

Process the event through error handling middleware.

Source code in src/aiogram_sentinel/middlewares/errors.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
async def __call__(
    self,
    handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
    event: TelegramObject,
    data: dict[str, Any],
) -> Any:
    """Process the event through error handling middleware."""
    try:
        return await handler(event, data)
    except Exception as exc:
        await self._handle_error(exc, event, data)
        # Don't re-raise - error has been handled
        return None

Policy Registry

aiogram_sentinel.PolicyRegistry 

PolicyRegistry()

Registry for managing named policies.

Initialize empty registry.

Source code in src/aiogram_sentinel/policy.py 
74
75
76
def __init__(self) -> None:
    """Initialize empty registry."""
    self._policies: OrderedDict[str, Policy] = OrderedDict()

register 

register(policy)

Register a new policy.

Parameters:

Name Type Description Default
policy Policy

Policy to register

 required

Raises:

Type Description
ValueError

If policy name already exists
Source code in src/aiogram_sentinel/policy.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
def register(self, policy: Policy) -> None:
    """Register a new policy.

    Args:
        policy: Policy to register

    Raises:
        ValueError: If policy name already exists
    """
    if policy.name in self._policies:
        raise ValueError(f"Policy '{policy.name}' already registered")

    self._policies[policy.name] = policy

get 

get(name)

Get policy by name.

Parameters:

Name Type Description Default
name str

Policy name

 required

Returns:

Type Description
Policy

Policy instance

Raises:

Type Description
ValueError

If policy not found, with suggestions
Source code in src/aiogram_sentinel/policy.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def get(self, name: str) -> Policy:
    """Get policy by name.

    Args:
        name: Policy name

    Returns:
        Policy instance

    Raises:
        ValueError: If policy not found, with suggestions
    """
    if name in self._policies:
        return self._policies[name]

    # Generate suggestions
    suggestions = difflib.get_close_matches(
        name, self._policies.keys(), n=3, cutoff=0.6
    )

    if suggestions:
        suggestion_text = f" Did you mean: {', '.join(suggestions)}?"
    else:
        suggestion_text = ""

    raise ValueError(f"Policy '{name}' not found.{suggestion_text}")

all 

all()

Get all registered policies in registration order.

Returns:

Type Description
list[Policy]

List of all policies
Source code in src/aiogram_sentinel/policy.py 
119
120
121
122
123
124
125
def all(self) -> list[Policy]:
    """Get all registered policies in registration order.

    Returns:
        List of all policies
    """
    return list(self._policies.values())

clear 

clear()

Clear all registered policies.

Source code in src/aiogram_sentinel/policy.py 
127
128
129
def clear(self) -> None:
    """Clear all registered policies."""
    self._policies.clear()

aiogram_sentinel.Policy dataclass 

Policy(name, kind, cfg, description='')

A named policy configuration.

name instance-attribute 

name

kind instance-attribute 

kind

cfg instance-attribute 

cfg

description class-attribute instance-attribute 

description = ''

aiogram_sentinel.ThrottleCfg dataclass 

ThrottleCfg(rate, per, scope=None, method=None, bucket=None)

Configuration for throttling policy.

rate instance-attribute 

rate

per instance-attribute 

per

scope class-attribute instance-attribute 

scope = None

method class-attribute instance-attribute 

method = None

bucket class-attribute instance-attribute 

bucket = None

aiogram_sentinel.DebounceCfg dataclass 

DebounceCfg(window, scope=None, method=None, bucket=None)

Configuration for debouncing policy.

window instance-attribute 

window

scope class-attribute instance-attribute 

scope = None

method class-attribute instance-attribute 

method = None

bucket class-attribute instance-attribute 

bucket = None

aiogram_sentinel.policy

Policy registry and configuration for aiogram-sentinel.

DebounceCfg dataclass 

DebounceCfg(window, scope=None, method=None, bucket=None)

Configuration for debouncing policy.

__post_init__ 

__post_init__()

Validate configuration after initialization.

Source code in src/aiogram_sentinel/policy.py 
44
45
46
47
def __post_init__(self) -> None:
    """Validate configuration after initialization."""
    if self.window <= 0:
        raise ValueError("window must be positive")

Policy dataclass 

Policy(name, kind, cfg, description='')

A named policy configuration.

__post_init__ 

__post_init__()

Validate policy after initialization.

Source code in src/aiogram_sentinel/policy.py 
59
60
61
62
63
64
65
66
67
68
def __post_init__(self) -> None:
    """Validate policy after initialization."""
    if not self.name:
        raise ValueError("policy name cannot be empty")

    # Validate cfg matches kind
    if self.kind == "throttle" and not isinstance(self.cfg, ThrottleCfg):
        raise ValueError("throttle policy must use ThrottleCfg")
    elif self.kind == "debounce" and not isinstance(self.cfg, DebounceCfg):
        raise ValueError("debounce policy must use DebounceCfg")

PolicyRegistry 

PolicyRegistry()

Registry for managing named policies.

Initialize empty registry.

Source code in src/aiogram_sentinel/policy.py 
74
75
76
def __init__(self) -> None:
    """Initialize empty registry."""
    self._policies: OrderedDict[str, Policy] = OrderedDict()

all 

all()

Get all registered policies in registration order.

Returns:

Type Description
list[Policy]

List of all policies
Source code in src/aiogram_sentinel/policy.py 
119
120
121
122
123
124
125
def all(self) -> list[Policy]:
    """Get all registered policies in registration order.

    Returns:
        List of all policies
    """
    return list(self._policies.values())

clear 

clear()

Clear all registered policies.

Source code in src/aiogram_sentinel/policy.py 
127
128
129
def clear(self) -> None:
    """Clear all registered policies."""
    self._policies.clear()

get 

get(name)

Get policy by name.

Parameters:

Name Type Description Default
name str

Policy name

 required

Returns:

Type Description
Policy

Policy instance

Raises:

Type Description
ValueError

If policy not found, with suggestions
Source code in src/aiogram_sentinel/policy.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def get(self, name: str) -> Policy:
    """Get policy by name.

    Args:
        name: Policy name

    Returns:
        Policy instance

    Raises:
        ValueError: If policy not found, with suggestions
    """
    if name in self._policies:
        return self._policies[name]

    # Generate suggestions
    suggestions = difflib.get_close_matches(
        name, self._policies.keys(), n=3, cutoff=0.6
    )

    if suggestions:
        suggestion_text = f" Did you mean: {', '.join(suggestions)}?"
    else:
        suggestion_text = ""

    raise ValueError(f"Policy '{name}' not found.{suggestion_text}")

register 

register(policy)

Register a new policy.

Parameters:

Name Type Description Default
policy Policy

Policy to register

 required

Raises:

Type Description
ValueError

If policy name already exists
Source code in src/aiogram_sentinel/policy.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
def register(self, policy: Policy) -> None:
    """Register a new policy.

    Args:
        policy: Policy to register

    Raises:
        ValueError: If policy name already exists
    """
    if policy.name in self._policies:
        raise ValueError(f"Policy '{policy.name}' already registered")

    self._policies[policy.name] = policy

ThrottleCfg dataclass 

ThrottleCfg(rate, per, scope=None, method=None, bucket=None)

Configuration for throttling policy.

__post_init__ 

__post_init__()

Validate configuration after initialization.

Source code in src/aiogram_sentinel/policy.py 
27
28
29
30
31
32
def __post_init__(self) -> None:
    """Validate configuration after initialization."""
    if self.rate <= 0:
        raise ValueError("rate must be positive")
    if self.per <= 0:
        raise ValueError("per must be positive")

coerce_scope 

coerce_scope(scope)

Coerce string scope to Scope enum with deprecation warning.

Parameters:

Name Type Description Default
scope str | Scope | None

String, Scope enum, or None

 required

Returns:

Type Description
Scope | None

Scope enum or None

Raises:

Type Description
ValueError

If string scope is invalid
Source code in src/aiogram_sentinel/policy.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def coerce_scope(scope: str | Scope | None) -> Scope | None:
    """Coerce string scope to Scope enum with deprecation warning.

    Args:
        scope: String, Scope enum, or None

    Returns:
        Scope enum or None

    Raises:
        ValueError: If string scope is invalid
    """
    if isinstance(scope, str):
        warnings.warn(
            "String scope is deprecated, use Scope enum",
            DeprecationWarning,
            stacklevel=3,
        )
        try:
            return Scope[scope.upper()]
        except KeyError:
            raise ValueError(f"Invalid scope: {scope}") from None

    return scope

convert_from_legacy_debounce 

convert_from_legacy_debounce(config)

Convert legacy debouncing config to DebounceCfg.

Parameters:

Name Type Description Default
config tuple[Any, ...] | dict[str, Any]

Legacy config tuple or dict

 required

Returns:

Type Description
DebounceCfg

DebounceCfg instance
Source code in src/aiogram_sentinel/policy.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def convert_from_legacy_debounce(
    config: tuple[Any, ...] | dict[str, Any],
) -> DebounceCfg:
    """Convert legacy debouncing config to DebounceCfg.

    Args:
        config: Legacy config tuple or dict

    Returns:
        DebounceCfg instance
    """
    if isinstance(config, (tuple, list)) and len(config) >= 1:
        window = config[0]
        scope_str = config[1] if len(config) > 1 else None
        scope = coerce_scope(scope_str) if scope_str else None
        return DebounceCfg(window=window, scope=scope)
    elif isinstance(config, dict):
        window = config.get("delay", config.get("window", 2))
        scope_str = config.get("scope")
        scope = coerce_scope(scope_str) if scope_str else None
        method = config.get("method")
        bucket = config.get("bucket")
        return DebounceCfg(window=window, scope=scope, method=method, bucket=bucket)
    else:
        raise ValueError(f"Invalid legacy debouncing config: {config}")

convert_from_legacy_throttle 

convert_from_legacy_throttle(config)

Convert legacy throttling config to ThrottleCfg.

Parameters:

Name Type Description Default
config tuple[Any, ...] | dict[str, Any]

Legacy config tuple or dict

 required

Returns:

Type Description
ThrottleCfg

ThrottleCfg instance
Source code in src/aiogram_sentinel/policy.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
def convert_from_legacy_throttle(
    config: tuple[Any, ...] | dict[str, Any],
) -> ThrottleCfg:
    """Convert legacy throttling config to ThrottleCfg.

    Args:
        config: Legacy config tuple or dict

    Returns:
        ThrottleCfg instance
    """
    if isinstance(config, (tuple, list)) and len(config) >= 2:
        rate, per = config[0], config[1]
        scope_str = config[2] if len(config) > 2 else None
        scope = coerce_scope(scope_str) if scope_str else None
        return ThrottleCfg(rate=rate, per=per, scope=scope)
    elif isinstance(config, dict):
        rate = config.get("limit", config.get("rate", 5))
        per = config.get("window", config.get("per", 10))
        scope_str = config.get("scope")
        scope = coerce_scope(scope_str) if scope_str else None
        method = config.get("method")
        bucket = config.get("bucket")
        return ThrottleCfg(
            rate=rate, per=per, scope=scope, method=method, bucket=bucket
        )
    else:
        raise ValueError(f"Invalid legacy throttling config: {config}")

policy 

policy(*names)

Decorator to attach policies to handlers.

Parameters:

Name Type Description Default
*names str

Policy names to attach

 ()

Returns:

Type Description
Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator function

Raises:

Type Description
ValueError

If no policy names provided
Source code in src/aiogram_sentinel/policy.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def policy(*names: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to attach policies to handlers.

    Args:
        *names: Policy names to attach

    Returns:
        Decorator function

    Raises:
        ValueError: If no policy names provided
    """
    if not names:
        raise ValueError("At least one policy name must be provided")

    def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
        """Attach policies to handler."""
        handler.__sentinel_policies__ = names  # type: ignore
        return handler

    return decorator

resolve_scope 

resolve_scope(user_id, chat_id, cap)

Resolve scope with cap constraint.

Parameters:

Name Type Description Default
user_id int | None

User identifier

 required
chat_id int | None

Chat identifier

 required
cap Scope | None

Maximum scope allowed (cap constraint)

 required

Returns:

Type Description
Scope | None

Resolved scope or None if cannot satisfy cap
Source code in src/aiogram_sentinel/policy.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def resolve_scope(
    user_id: int | None, chat_id: int | None, cap: Scope | None
) -> Scope | None:
    """Resolve scope with cap constraint.

    Args:
        user_id: User identifier
        chat_id: Chat identifier
        cap: Maximum scope allowed (cap constraint)

    Returns:
        Resolved scope or None if cannot satisfy cap
    """
    # Determine available scopes
    available: set[Scope] = set()
    if user_id and chat_id:
        available.add(Scope.GROUP)
    if user_id:
        available.add(Scope.USER)
    if chat_id:
        available.add(Scope.CHAT)
    available.add(Scope.GLOBAL)

    # Specificity order (most specific first)
    order = [Scope.USER, Scope.CHAT, Scope.GROUP, Scope.GLOBAL]

    # Filter by cap if provided
    if cap:
        cap_idx = order.index(cap)
        candidates = order[: cap_idx + 1]
    else:
        candidates = order

    # Return first available in specificity order
    for s in candidates:
        if s in available:
            return s

    return None

aiogram_sentinel.coerce_scope 

coerce_scope(scope)

Coerce string scope to Scope enum with deprecation warning.

Parameters:

Name Type Description Default
scope str | Scope | None

String, Scope enum, or None

 required

Returns:

Type Description
Scope | None

Scope enum or None

Raises:

Type Description
ValueError

If string scope is invalid
Source code in src/aiogram_sentinel/policy.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def coerce_scope(scope: str | Scope | None) -> Scope | None:
    """Coerce string scope to Scope enum with deprecation warning.

    Args:
        scope: String, Scope enum, or None

    Returns:
        Scope enum or None

    Raises:
        ValueError: If string scope is invalid
    """
    if isinstance(scope, str):
        warnings.warn(
            "String scope is deprecated, use Scope enum",
            DeprecationWarning,
            stacklevel=3,
        )
        try:
            return Scope[scope.upper()]
        except KeyError:
            raise ValueError(f"Invalid scope: {scope}") from None

    return scope

aiogram_sentinel.resolve_scope 

resolve_scope(user_id, chat_id, cap)

Resolve scope with cap constraint.

Parameters:

Name Type Description Default
user_id int | None

User identifier

 required
chat_id int | None

Chat identifier

 required
cap Scope | None

Maximum scope allowed (cap constraint)

 required

Returns:

Type Description
Scope | None

Resolved scope or None if cannot satisfy cap
Source code in src/aiogram_sentinel/policy.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def resolve_scope(
    user_id: int | None, chat_id: int | None, cap: Scope | None
) -> Scope | None:
    """Resolve scope with cap constraint.

    Args:
        user_id: User identifier
        chat_id: Chat identifier
        cap: Maximum scope allowed (cap constraint)

    Returns:
        Resolved scope or None if cannot satisfy cap
    """
    # Determine available scopes
    available: set[Scope] = set()
    if user_id and chat_id:
        available.add(Scope.GROUP)
    if user_id:
        available.add(Scope.USER)
    if chat_id:
        available.add(Scope.CHAT)
    available.add(Scope.GLOBAL)

    # Specificity order (most specific first)
    order = [Scope.USER, Scope.CHAT, Scope.GROUP, Scope.GLOBAL]

    # Filter by cap if provided
    if cap:
        cap_idx = order.index(cap)
        candidates = order[: cap_idx + 1]
    else:
        candidates = order

    # Return first available in specificity order
    for s in candidates:
        if s in available:
            return s

    return None

Main Classes

aiogram_sentinel.Sentinel

Main setup class for aiogram-sentinel.

setup async staticmethod 

setup(dp, cfg, router=None, *, infra=None, error_config=None)

Setup aiogram-sentinel middlewares.

Parameters:

Name Type Description Default
dp Dispatcher

Dispatcher instance

 required
cfg SentinelConfig

Configuration

 required
router Router | None

Optional router to use (creates new one if not provided)

 None
infra InfraBundle | None

Optional infrastructure bundle (builds from config if not provided)

 None
error_config ErrorConfig | None

Optional error handling configuration

 None

Returns:

Type Description
tuple[Router, InfraBundle]

Tuple of (router, infra_bundle)
Source code in src/aiogram_sentinel/sentinel.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@staticmethod
async def setup(
    dp: Dispatcher,
    cfg: SentinelConfig,
    router: Router | None = None,
    *,
    infra: InfraBundle | None = None,
    error_config: ErrorConfig | None = None,
) -> tuple[Router, InfraBundle]:
    """Setup aiogram-sentinel middlewares.

    Args:
        dp: Dispatcher instance
        cfg: Configuration
        router: Optional router to use (creates new one if not provided)
        infra: Optional infrastructure bundle (builds from config if not provided)
        error_config: Optional error handling configuration

    Returns:
        Tuple of (router, infra_bundle)
    """
    # Build infrastructure if not provided
    if infra is None:
        infra = build_infra(cfg)

    # Create KeyBuilder instance
    key_builder = KeyBuilder(app=cfg.redis_prefix)

    # Create or use provided router
    if router is None:
        router = Router(name="sentinel")

    # Create middlewares in correct order with KeyBuilder
    policy_resolver = PolicyResolverMiddleware(registry, cfg)
    debounce_middleware = DebounceMiddleware(infra.debounce, cfg, key_builder)
    throttling_middleware = ThrottlingMiddleware(
        infra.rate_limiter, cfg, key_builder
    )

    # Create error handling middleware if configured
    error_middleware = None
    effective_error_config = error_config or Sentinel._error_config
    if effective_error_config:
        error_middleware = ErrorHandlingMiddleware(
            effective_error_config, key_builder, infra.rate_limiter
        )

    # Add middlewares to router in correct order
    for reg in (router.message, router.callback_query):
        if error_middleware:
            reg.middleware(error_middleware)  # FIRST (outermost)
        reg.middleware(policy_resolver)
        reg.middleware(debounce_middleware)
        reg.middleware(throttling_middleware)

    # Include router in dispatcher
    dp.include_router(router)

    return router, infra

add_hooks staticmethod 

add_hooks(router, infra, cfg, *, on_rate_limited=None)

Add hooks to existing middlewares.

Parameters:

Name Type Description Default
router Router

Router with middlewares

 required
infra InfraBundle

Infrastructure bundle (rate_limiter, debounce)

 required
cfg SentinelConfig

SentinelConfig configuration

 required
on_rate_limited Callable[[Any, dict[str, Any], float], Awaitable[Any]] | None

Optional hook for rate-limited events

 None
Source code in src/aiogram_sentinel/sentinel.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@staticmethod
def add_hooks(
    router: Router,
    infra: InfraBundle,
    cfg: SentinelConfig,
    *,
    on_rate_limited: Callable[[Any, dict[str, Any], float], Awaitable[Any]]
    | None = None,
) -> None:
    """Add hooks to existing middlewares.

    Args:
        router: Router with middlewares
        infra: Infrastructure bundle (rate_limiter, debounce)
        cfg: SentinelConfig configuration
        on_rate_limited: Optional hook for rate-limited events
    """
    # Create KeyBuilder instance
    key_builder = KeyBuilder(app=cfg.redis_prefix)

    # Create middlewares with hooks
    policy_resolver = PolicyResolverMiddleware(registry, cfg)
    debounce_middleware = DebounceMiddleware(infra.debounce, cfg, key_builder)
    throttling_middleware = ThrottlingMiddleware(
        infra.rate_limiter, cfg, key_builder, on_rate_limited=on_rate_limited
    )

    # Replace middlewares with hook-enabled versions
    for reg in (router.message, router.callback_query):
        # Clear existing middlewares
        reg.middlewares.clear()  # type: ignore

        # Add complete middleware chain with hooks in correct order
        reg.middleware(policy_resolver)  # FIRST
        reg.middleware(debounce_middleware)
        reg.middleware(throttling_middleware)

aiogram_sentinel.SentinelConfig dataclass 

SentinelConfig(backend='memory', redis_url='redis://localhost:6379', redis_prefix='sentinel', throttling_default_max=5, throttling_default_per_seconds=10, debounce_default_window=2, require_registration=False)

Configuration for aiogram-sentinel.

throttling_default_max class-attribute instance-attribute 

throttling_default_max = 5

throttling_default_per_seconds class-attribute instance-attribute 

throttling_default_per_seconds = 10

debounce_default_window class-attribute instance-attribute 

debounce_default_window = 2

backend class-attribute instance-attribute 

backend = 'memory'

redis_url class-attribute instance-attribute 

redis_url = 'redis://localhost:6379'

redis_prefix class-attribute instance-attribute 

redis_prefix = 'sentinel'

aiogram_sentinel.InfraBundle dataclass 

InfraBundle(rate_limiter, debounce)

Bundle of infrastructure backends managed by the library.

aiogram_sentinel.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."""
    ...

aiogram_sentinel.KeyBuilder 

KeyBuilder(app, sep=':')

Composite key builder with collision-proof scheme.

Initialize KeyBuilder.

Parameters:

Name Type Description Default
app str

Application prefix (typically from redis_prefix config)

 required
sep str

Key separator character

 ':'
Source code in src/aiogram_sentinel/scopes.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(self, app: str, sep: str = ":") -> None:
    """Initialize KeyBuilder.

    Args:
        app: Application prefix (typically from redis_prefix config)
        sep: Key separator character
    """
    if not app:
        raise ValueError("app cannot be empty")
    if sep in app:
        raise ValueError(f"app cannot contain separator '{sep}': {app}")

    self.app = app
    self.sep = sep

for_update 

for_update(parts, *, method=None, bucket=None)

Build canonical key for storage/metrics with stable ordering.

Parameters:

Name Type Description Default
parts KeyParts

Key parts containing namespace, scope, and identifiers

 required
method str | None

Optional method name (e.g., "sendMessage")

 None
bucket str | None

Optional bucket identifier

 None

Returns:

Type Description
str

Canonical key string

Raises:

Type Description
ValueError

If method or bucket contain separator characters
Source code in src/aiogram_sentinel/scopes.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def for_update(
    self,
    parts: KeyParts,
    *,
    method: str | None = None,
    bucket: str | None = None,
) -> str:
    """Build canonical key for storage/metrics with stable ordering.

    Args:
        parts: Key parts containing namespace, scope, and identifiers
        method: Optional method name (e.g., "sendMessage")
        bucket: Optional bucket identifier

    Returns:
        Canonical key string

    Raises:
        ValueError: If method or bucket contain separator characters
    """
    # Validate optional parameters
    if method is not None and self.sep in method:
        raise ValueError(f"method cannot contain separator '{self.sep}': {method}")
    if bucket is not None and self.sep in bucket:
        raise ValueError(f"bucket cannot contain separator '{self.sep}': {bucket}")

    # Build key components
    key_parts = [self.app, parts.namespace, parts.scope.name]

    # Add identifiers
    key_parts.extend(parts.identifiers)

    # Add optional method parameter
    if method is not None:
        key_parts.append(f"m={method}")

    # Add optional bucket parameter
    if bucket is not None:
        key_parts.append(f"b={bucket}")

    return self.sep.join(key_parts)

user 

user(namespace, user_id, **kwargs)

Build key for user scope.

Parameters:

Name Type Description Default
namespace str

Key namespace (e.g., "throttle", "debounce")

 required
user_id int

User identifier

 required
**kwargs Any

Additional parameters (method, bucket)

 {}

Returns:

Type Description
str

User-scoped key
Source code in src/aiogram_sentinel/scopes.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def user(self, namespace: str, user_id: int, **kwargs: Any) -> str:
    """Build key for user scope.

    Args:
        namespace: Key namespace (e.g., "throttle", "debounce")
        user_id: User identifier
        **kwargs: Additional parameters (method, bucket)

    Returns:
        User-scoped key
    """
    parts = KeyParts(
        namespace=namespace, scope=Scope.USER, identifiers=(str(user_id),)
    )
    return self.for_update(parts, **kwargs)

chat 

chat(namespace, chat_id, **kwargs)

Build key for chat scope.

Parameters:

Name Type Description Default
namespace str

Key namespace (e.g., "throttle", "debounce")

 required
chat_id int

Chat identifier

 required
**kwargs Any

Additional parameters (method, bucket)

 {}

Returns:

Type Description
str

Chat-scoped key
Source code in src/aiogram_sentinel/scopes.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def chat(self, namespace: str, chat_id: int, **kwargs: Any) -> str:
    """Build key for chat scope.

    Args:
        namespace: Key namespace (e.g., "throttle", "debounce")
        chat_id: Chat identifier
        **kwargs: Additional parameters (method, bucket)

    Returns:
        Chat-scoped key
    """
    parts = KeyParts(
        namespace=namespace, scope=Scope.CHAT, identifiers=(str(chat_id),)
    )
    return self.for_update(parts, **kwargs)

group 

group(namespace, user_id, chat_id, **kwargs)

Build key for group scope (user+chat composite).

Parameters:

Name Type Description Default
namespace str

Key namespace (e.g., "throttle", "debounce")

 required
user_id int

User identifier

 required
chat_id int

Chat identifier

 required
**kwargs Any

Additional parameters (method, bucket)

 {}

Returns:

Type Description
str

Group-scoped key
Source code in src/aiogram_sentinel/scopes.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def group(self, namespace: str, user_id: int, chat_id: int, **kwargs: Any) -> str:
    """Build key for group scope (user+chat composite).

    Args:
        namespace: Key namespace (e.g., "throttle", "debounce")
        user_id: User identifier
        chat_id: Chat identifier
        **kwargs: Additional parameters (method, bucket)

    Returns:
        Group-scoped key
    """
    parts = KeyParts(
        namespace=namespace,
        scope=Scope.GROUP,
        identifiers=(str(user_id), str(chat_id)),
    )
    return self.for_update(parts, **kwargs)

global_ 

global_(namespace, **kwargs)

Build key for global scope.

Parameters:

Name Type Description Default
namespace str

Key namespace (e.g., "throttle", "debounce")

 required
**kwargs Any

Additional parameters (method, bucket)

 {}

Returns:

Type Description
str

Global-scoped key
Source code in src/aiogram_sentinel/scopes.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def global_(self, namespace: str, **kwargs: Any) -> str:
    """Build key for global scope.

    Args:
        namespace: Key namespace (e.g., "throttle", "debounce")
        **kwargs: Additional parameters (method, bucket)

    Returns:
        Global-scoped key
    """
    parts = KeyParts(
        namespace=namespace, scope=Scope.GLOBAL, identifiers=("global",)
    )
    return self.for_update(parts, **kwargs)

aiogram_sentinel.KeyParts dataclass 

KeyParts(namespace, scope, identifiers)

Immutable key parts for consistent key generation.

namespace instance-attribute 

namespace

scope instance-attribute 

scope

identifiers instance-attribute 

identifiers

aiogram_sentinel.Scope

Bases: Enum

Scope enumeration for key generation.

USER class-attribute instance-attribute 

USER = 'user'

CHAT class-attribute instance-attribute 

CHAT = 'chat'

GROUP class-attribute instance-attribute 

GROUP = 'group'

GLOBAL class-attribute instance-attribute 

GLOBAL = 'global'

Context Extractors

aiogram_sentinel.context.extract_user_id 

extract_user_id(event, data)

Extract user ID from event with fallbacks.

Parameters:

Name Type Description Default
event TelegramObject

Telegram event object

 required
data dict[str, Any]

Middleware data dictionary

 required

Returns:

Type Description
int | None

User ID if found, None otherwise
Source code in src/aiogram_sentinel/context.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def extract_user_id(event: TelegramObject, data: dict[str, Any]) -> int | None:
    """Extract user ID from event with fallbacks.

    Args:
        event: Telegram event object
        data: Middleware data dictionary

    Returns:
        User ID if found, None otherwise
    """
    # Try from_user attribute (most common)
    if hasattr(event, "from_user") and getattr(event, "from_user", None):  # type: ignore[attr-defined]
        return getattr(event.from_user, "id", None)  # type: ignore[attr-defined]

    # Try user attribute (some event types)
    if hasattr(event, "user") and getattr(event, "user", None):  # type: ignore[attr-defined]
        return getattr(event.user, "id", None)  # type: ignore[attr-defined]

    # Try chat attribute as fallback (for anonymous events)
    if hasattr(event, "chat") and getattr(event, "chat", None):  # type: ignore[attr-defined]
        chat_id = getattr(event.chat, "id", None)  # type: ignore[attr-defined]
        # Only return chat ID if it's a private chat (user ID == chat ID)
        # and it's positive (not a group/supergroup)
        if chat_id and chat_id > 0 and getattr(event.chat, "type", None) == "private":  # type: ignore[attr-defined]
            return chat_id

    return None

aiogram_sentinel.context.extract_chat_id 

extract_chat_id(event, data)

Extract chat ID from event.

Parameters:

Name Type Description Default
event TelegramObject

Telegram event object

 required
data dict[str, Any]

Middleware data dictionary

 required

Returns:

Type Description
int | None

Chat ID if found, None otherwise
Source code in src/aiogram_sentinel/context.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def extract_chat_id(event: TelegramObject, data: dict[str, Any]) -> int | None:
    """Extract chat ID from event.

    Args:
        event: Telegram event object
        data: Middleware data dictionary

    Returns:
        Chat ID if found, None otherwise
    """
    # Try chat attribute
    if hasattr(event, "chat") and getattr(event, "chat", None):  # type: ignore[attr-defined]
        return getattr(event.chat, "id", None)  # type: ignore[attr-defined]

    # Try message attribute (for some event types)
    if hasattr(event, "message") and getattr(event, "message", None):
        message = getattr(event, "message", None)
        if message and hasattr(message, "chat") and getattr(message, "chat", None):
            return getattr(message.chat, "id", None)

    return None

aiogram_sentinel.context.extract_group_ids 

extract_group_ids(event, data)

Extract both user and chat IDs for group scope.

Parameters:

Name Type Description Default
event TelegramObject

Telegram event object

 required
data dict[str, Any]

Middleware data dictionary

 required

Returns:

Type Description
tuple[int | None, int | None]

Tuple of (user_id, chat_id)
Source code in src/aiogram_sentinel/context.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def extract_group_ids(
    event: TelegramObject, data: dict[str, Any]
) -> tuple[int | None, int | None]:
    """Extract both user and chat IDs for group scope.

    Args:
        event: Telegram event object
        data: Middleware data dictionary

    Returns:
        Tuple of (user_id, chat_id)
    """
    user_id = extract_user_id(event, data)
    chat_id = extract_chat_id(event, data)
    return user_id, chat_id

aiogram_sentinel.context.extract_event_type 

extract_event_type(event, data)

Extract event type for classification.

Parameters:

Name Type Description Default
event TelegramObject

Telegram event object

 required
data dict[str, Any]

Middleware data dictionary

 required

Returns:

Type Description
str

Event type string
Source code in src/aiogram_sentinel/context.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def extract_event_type(event: TelegramObject, data: dict[str, Any]) -> str:
    """Extract event type for classification.

    Args:
        event: Telegram event object
        data: Middleware data dictionary

    Returns:
        Event type string
    """
    # Get class name and convert to lowercase
    event_type = event.__class__.__name__.lower()

    # Map specific event types to more readable names
    type_mapping = {
        "message": "message",
        "callbackquery": "callback",
        "inlinequery": "inline",
        "choseninlineresult": "inline",
        "chatmemberupdated": "chat_member",
        "mycommand": "command",
        "chatjoinrequest": "join_request",
        "chatboostupdated": "boost",
        "chatboostremoved": "boost",
        "messageautodeletetimerchanged": "auto_delete",
        "forumtopiccreated": "forum_topic",
        "forumtopicclosed": "forum_topic",
        "forumtopicreopened": "forum_topic",
        "forumtopicedited": "forum_topic",
        "generalforumtopichidden": "forum_topic",
        "generalforumtopicunhidden": "forum_topic",
        "forumtopicpinned": "forum_topic",
        "forumtopicunpinned": "forum_topic",
        "writeaccessallowed": "write_access",
        "userprofilephotos": "profile_photos",
        "usershared": "user_shared",
        "chatshared": "chat_shared",
        "story": "story",
        "storydeleted": "story",
        "videonote": "video_note",
        "voice": "voice",
        "video": "video",
        "photo": "photo",
        "document": "document",
        "animation": "animation",
        "sticker": "sticker",
        "contact": "contact",
        "location": "location",
        "venue": "venue",
        "poll": "poll",
        "dice": "dice",
        "game": "game",
        "invoice": "invoice",
        "successfulpayment": "payment",
        "passportdata": "passport",
        "proximityalerttriggered": "proximity",
        "webappdata": "webapp",
        "videochatstarted": "video_chat",
        "videochatended": "video_chat",
        "videochatparticipantsinvited": "video_chat",
        "videochatscheduled": "video_chat",
        "webapp": "webapp",
        "giveawaycreated": "giveaway",
        "giveaway": "giveaway",
        "giveawaywinners": "giveaway",
        "giveawaycompleted": "giveaway",
        "businessconnection": "business",
        "businessmessagesdeleted": "business",
        "businessintro": "business",
        "businesslocation": "business",
        "businessopeninghours": "business",
        "businessawaymessage": "business",
        "businessgreetingmessage": "business",
        "businesschat": "business",
    }

    return type_mapping.get(event_type, event_type)

aiogram_sentinel.context.extract_handler_bucket 

extract_handler_bucket(event, data)

Extract handler bucket from event and data.

Parameters:

Name Type Description Default
event TelegramObject

Telegram event object

 required
data dict[str, Any]

Middleware data dictionary

 required

Returns:

Type Description
str | None

Handler bucket string if found, None otherwise
Source code in src/aiogram_sentinel/context.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def extract_handler_bucket(event: TelegramObject, data: dict[str, Any]) -> str | None:
    """Extract handler bucket from event and data.

    Args:
        event: Telegram event object
        data: Middleware data dictionary

    Returns:
        Handler bucket string if found, None otherwise
    """
    # Try to get handler name from data
    if "handler" in data:
        handler = data["handler"]
        if hasattr(handler, "__name__"):
            return handler.__name__

    # Try to get handler from event attributes
    if hasattr(event, "handler") and getattr(event, "handler", None):
        handler = getattr(event, "handler", None)
        if handler and hasattr(handler, "__name__"):
            return handler.__name__

    return None

aiogram_sentinel.context.extract_callback_bucket 

extract_callback_bucket(event, data)

Extract callback bucket from callback query events.

Parameters:

Name Type Description Default
event TelegramObject

Telegram event object

 required
data dict[str, Any]

Middleware data dictionary

 required

Returns:

Type Description
str | None

Callback bucket string if found, None otherwise
Source code in src/aiogram_sentinel/context.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def extract_callback_bucket(event: TelegramObject, data: dict[str, Any]) -> str | None:
    """Extract callback bucket from callback query events.

    Args:
        event: Telegram event object
        data: Middleware data dictionary

    Returns:
        Callback bucket string if found, None otherwise
    """
    # Only process callback query events
    if not hasattr(event, "data"):
        return None

    callback_data = getattr(event, "data", None)
    if callback_data is None:
        return None

    # Convert to string if not already
    callback_data = str(callback_data)

    # Handle empty string case
    if not callback_data:
        return ""

    # Parse callback data to extract action
    # Common patterns: "action:param", "action_param", "action"
    if ":" in callback_data:
        return callback_data.split(":", 1)[0]
    elif "_" in callback_data and len(callback_data.split("_")) > 2:
        # Only split on underscore if there are multiple parts
        return callback_data.split("_", 1)[0]
    else:
        return callback_data

Middleware

aiogram_sentinel.middlewares.ThrottlingMiddleware 

ThrottlingMiddleware(rate_limiter, cfg, key_builder, on_rate_limited=None)

Bases: BaseMiddleware

Middleware for rate limiting with optional notifier hook.

Initialize the throttling middleware.

Parameters:

Name Type Description Default
rate_limiter RateLimiterBackend

Rate limiter backend instance

 required
cfg SentinelConfig

SentinelConfig configuration

 required
key_builder KeyBuilder

KeyBuilder instance for key generation

 required
on_rate_limited Callable[[TelegramObject, dict[str, Any], float], Awaitable[None]] | None

Optional hook called when rate limit is exceeded

 None
Source code in src/aiogram_sentinel/middlewares/throttling.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    rate_limiter: RateLimiterBackend,
    cfg: SentinelConfig,
    key_builder: KeyBuilder,
    on_rate_limited: Callable[
        [TelegramObject, dict[str, Any], float], Awaitable[None]
    ]
    | None = None,
) -> None:
    """Initialize the throttling middleware.

    Args:
        rate_limiter: Rate limiter backend instance
        cfg: SentinelConfig configuration
        key_builder: KeyBuilder instance for key generation
        on_rate_limited: Optional hook called when rate limit is exceeded
    """
    super().__init__()
    self._rate_limiter = rate_limiter
    self._cfg = cfg
    self._key_builder = key_builder
    self._on_rate_limited = on_rate_limited
    self._default_limit = cfg.throttling_default_max
    self._default_window = cfg.throttling_default_per_seconds

__call__ async 

__call__(handler, event, data)

Process the event through throttling middleware.

Source code in src/aiogram_sentinel/middlewares/throttling.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def __call__(
    self,
    handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
    event: TelegramObject,
    data: dict[str, Any],
) -> Any:
    """Process the event through throttling middleware."""
    # Get rate limit configuration from handler or use defaults
    max_events, per_seconds = self._get_rate_limit_config(handler, data, event)

    # Generate rate limit key
    key = self._generate_rate_limit_key(event, handler, data)

    # Check if request is allowed
    allowed = await self._rate_limiter.allow(key, max_events, per_seconds)

    if not allowed:
        # Rate limit exceeded
        data["sentinel_rate_limited"] = True

        # Calculate retry after time using actual remaining TTL
        retry_after = await self._calculate_retry_after(key, per_seconds)

        # Call optional hook
        if self._on_rate_limited:
            try:
                await self._on_rate_limited(event, data, retry_after)
            except Exception as e:
                logger.exception("on_rate_limited hook failed: %s", e)

        # Stop processing
        return

    # Continue to next middleware/handler
    return await handler(event, data)

aiogram_sentinel.middlewares.DebounceMiddleware 

DebounceMiddleware(debounce_backend, cfg, key_builder)

Bases: BaseMiddleware

Middleware for debouncing duplicate messages with fingerprinting.

Initialize the debouncing middleware.

Parameters:

Name Type Description Default
debounce_backend DebounceBackend

Debounce backend instance

 required
cfg SentinelConfig

SentinelConfig configuration

 required
key_builder KeyBuilder

KeyBuilder instance for key generation

 required
Source code in src/aiogram_sentinel/middlewares/debouncing.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    debounce_backend: DebounceBackend,
    cfg: SentinelConfig,
    key_builder: KeyBuilder,
) -> None:
    """Initialize the debouncing middleware.

    Args:
        debounce_backend: Debounce backend instance
        cfg: SentinelConfig configuration
        key_builder: KeyBuilder instance for key generation
    """
    super().__init__()
    self._debounce_backend = debounce_backend
    self._cfg = cfg
    self._key_builder = key_builder
    self._default_delay = cfg.debounce_default_window

__call__ async 

__call__(handler, event, data)

Process the event through debouncing middleware.

Source code in src/aiogram_sentinel/middlewares/debouncing.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def __call__(
    self,
    handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
    event: TelegramObject,
    data: dict[str, Any],
) -> Any:
    """Process the event through debouncing middleware."""
    # Get debounce configuration
    window_seconds = self._get_debounce_window(handler, data, event)

    # Generate fingerprint for the event
    fp = self._generate_fingerprint(event)

    # Generate debounce key
    key = self._generate_debounce_key(event, handler, data)

    # Check if already seen within window
    if await self._debounce_backend.seen(key, window_seconds, fp):
        # Duplicate detected within window
        data["sentinel_debounced"] = True
        return  # Stop processing

    # Continue to next middleware/handler
    return await handler(event, data)

Decorators

aiogram_sentinel.decorators.rate_limit 

rate_limit(max_events, per_seconds, *, scope=None)

Decorator to set rate limit configuration on handlers.

Parameters:

Name Type Description Default
max_events int

Maximum number of events per time window

 required
per_seconds int

Time window in seconds

 required
scope str | None

Optional scope for rate limiting

 None
Deprecated

Use @policy() with PolicyRegistry instead. Will be removed in v2.0.0.

Source code in src/aiogram_sentinel/decorators.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def rate_limit(
    max_events: int, per_seconds: int, *, scope: str | None = None
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to set rate limit configuration on handlers.

    Args:
        max_events: Maximum number of events per time window
        per_seconds: Time window in seconds
        scope: Optional scope for rate limiting

    Deprecated:
        Use @policy() with PolicyRegistry instead. Will be removed in v2.0.0.
    """

    def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
        warnings.warn(
            "@rate_limit is deprecated, use @policy() with PolicyRegistry instead. "
            "Will be removed in v2.0.0",
            DeprecationWarning,
            stacklevel=2,
        )
        # Store rate limit configuration on the handler
        handler.sentinel_rate_limit = (max_events, per_seconds, scope)  # type: ignore
        return handler

    return decorator

aiogram_sentinel.decorators.debounce 

debounce(window_seconds, *, scope=None)

Decorator to set debounce configuration on handlers.

Parameters:

Name Type Description Default
window_seconds int

Debounce window in seconds

 required
scope str | None

Optional scope for debouncing

 None
Deprecated

Use @policy() with PolicyRegistry instead. Will be removed in v2.0.0.

Source code in src/aiogram_sentinel/decorators.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def debounce(
    window_seconds: int, *, scope: str | None = None
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to set debounce configuration on handlers.

    Args:
        window_seconds: Debounce window in seconds
        scope: Optional scope for debouncing

    Deprecated:
        Use @policy() with PolicyRegistry instead. Will be removed in v2.0.0.
    """

    def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
        warnings.warn(
            "@debounce is deprecated, use @policy() with PolicyRegistry instead. "
            "Will be removed in v2.0.0",
            DeprecationWarning,
            stacklevel=2,
        )
        # Store debounce configuration on the handler
        handler.sentinel_debounce = (window_seconds, scope)  # type: ignore
        return handler

    return decorator

aiogram_sentinel.policy.policy 

policy(*names)

Decorator to attach policies to handlers.

Parameters:

Name Type Description Default
*names str

Policy names to attach

 ()

Returns:

Type Description
Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator function

Raises:

Type Description
ValueError

If no policy names provided
Source code in src/aiogram_sentinel/policy.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def policy(*names: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to attach policies to handlers.

    Args:
        *names: Policy names to attach

    Returns:
        Decorator function

    Raises:
        ValueError: If no policy names provided
    """
    if not names:
        raise ValueError("At least one policy name must be provided")

    def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:
        """Attach policies to handler."""
        handler.__sentinel_policies__ = names  # type: ignore
        return handler

    return decorator

Utilities

Key Generation

aiogram_sentinel.utils.keys.rate_key 

rate_key(user_id, handler_name, **kwargs)

Build rate limiting key from user ID and handler scope.

.. deprecated:: 1.1.0 Use :class:KeyBuilder instead. This function will be removed in v2.0.0.

Migration:
.. code-block:: python

    # Old
    key = rate_key(user_id, handler_name, **kwargs)

    # New
    from aiogram_sentinel import KeyBuilder
    kb = KeyBuilder(app="sentinel")
    key = kb.user("throttle", user_id, bucket=handler_name, **kwargs)
Source code in src/aiogram_sentinel/utils/keys.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def rate_key(user_id: int, handler_name: str, **kwargs: Any) -> str:
    """Build rate limiting key from user ID and handler scope.

    .. deprecated:: 1.1.0
        Use :class:`KeyBuilder` instead. This function will be removed in v2.0.0.

        Migration:
        .. code-block:: python

            # Old
            key = rate_key(user_id, handler_name, **kwargs)

            # New
            from aiogram_sentinel import KeyBuilder
            kb = KeyBuilder(app="sentinel")
            key = kb.user("throttle", user_id, bucket=handler_name, **kwargs)
    """
    # Emit deprecation warning
    warnings.warn(
        "rate_key() is deprecated and will be removed in v2.0.0. "
        "Use KeyBuilder instead.",
        DeprecationWarning,
        stacklevel=2,
    )

    # Log deprecation warning once
    logger.warning(
        "rate_key() is deprecated and will be removed in v2.0.0. "
        "Use KeyBuilder instead."
    )

    # Create a stable key from user_id and handler_name (original implementation)
    key_parts = [str(user_id), handler_name]

    # Add any additional scope parameters
    for key, value in sorted(kwargs.items()):
        key_parts.append(f"{key}:{value}")

    return ":".join(key_parts)

aiogram_sentinel.utils.keys.debounce_key 

debounce_key(user_id, handler_name, **kwargs)

Build debounce key from user ID and handler scope.

.. deprecated:: 1.1.0 Use :class:KeyBuilder instead. This function will be removed in v2.0.0.

Migration:
.. code-block:: python

    # Old
    key = debounce_key(user_id, handler_name, **kwargs)

    # New
    from aiogram_sentinel import KeyBuilder
    kb = KeyBuilder(app="sentinel")
    key = kb.user("debounce", user_id, bucket=handler_name, **kwargs)
Source code in src/aiogram_sentinel/utils/keys.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def debounce_key(user_id: int, handler_name: str, **kwargs: Any) -> str:
    """Build debounce key from user ID and handler scope.

    .. deprecated:: 1.1.0
        Use :class:`KeyBuilder` instead. This function will be removed in v2.0.0.

        Migration:
        .. code-block:: python

            # Old
            key = debounce_key(user_id, handler_name, **kwargs)

            # New
            from aiogram_sentinel import KeyBuilder
            kb = KeyBuilder(app="sentinel")
            key = kb.user("debounce", user_id, bucket=handler_name, **kwargs)
    """
    # Emit deprecation warning
    warnings.warn(
        "debounce_key() is deprecated and will be removed in v2.0.0. "
        "Use KeyBuilder instead.",
        DeprecationWarning,
        stacklevel=2,
    )

    # Log deprecation warning once
    logger.warning(
        "debounce_key() is deprecated and will be removed in v2.0.0. "
        "Use KeyBuilder instead."
    )

    # Create a stable key from user_id and handler_name (original implementation)
    key_parts = [str(user_id), handler_name]

    # Add any additional scope parameters
    for key, value in sorted(kwargs.items()):
        key_parts.append(f"{key}:{value}")

    return ":".join(key_parts)

aiogram_sentinel.utils.keys.fingerprint 

fingerprint(text)

Create a stable fingerprint for text content.

Source code in src/aiogram_sentinel/utils/keys.py 
111
112
113
114
115
116
117
118
119
def fingerprint(text: str | None) -> str:
    """Create a stable fingerprint for text content."""
    # Handle None, empty strings, and non-string types
    if not text:
        text = ""
    else:
        text = str(text)

    return hashlib.sha256(text.encode()).hexdigest()[:16]