Skip to content

Router

The router dispatches incoming callback queries to registered handler functions based on the decoded action.

telegram_menu_builder.router.MenuRouter

Router for dispatching callback queries to handlers.

This class manages handler registration and routes incoming callback queries to the appropriate handler function based on the decoded action.

Attributes:

Name Type Description
storage StorageBackend

Storage backend for callback data
encoder CallbackEncoder

Callback data encoder/decoder
handlers CallbackEncoder

Mapping of handler names to functions
default_handler CallbackEncoder

Fallback handler for unknown actions
Example

router = MenuRouter()

@router.handler("edit_user") async def handle_edit(update, context, params): ... user_id = params["user_id"] ... # Handle edit...

Register with application

app.add_handler(CallbackQueryHandler(router.route))

Source code in src/telegram_menu_builder/router.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
 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
156
157
158
159
160
161
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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
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
254
255
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
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
class MenuRouter:
    """Router for dispatching callback queries to handlers.

    This class manages handler registration and routes incoming callback queries
    to the appropriate handler function based on the decoded action.

    Attributes:
        storage: Storage backend for callback data
        encoder: Callback data encoder/decoder
        handlers: Mapping of handler names to functions
        default_handler: Fallback handler for unknown actions

    Example:
        >>> router = MenuRouter()
        >>>
        >>> @router.handler("edit_user")
        >>> async def handle_edit(update, context, params):
        ...     user_id = params["user_id"]
        ...     # Handle edit...
        >>>
        >>> # Register with application
        >>> app.add_handler(CallbackQueryHandler(router.route))
    """

    def __init__(
        self,
        storage: StorageBackend | None = None,
        default_handler: HandlerFunc | None = None,
        auto_answer: bool = True,
    ) -> None:
        """Initialize menu router.

        Args:
            storage: Storage backend (defaults to MemoryStorage)
            default_handler: Optional fallback handler for unknown actions
            auto_answer: Automatically answer callback queries
        """
        self._storage = storage or MemoryStorage()
        self._encoder = CallbackEncoder(self._storage)
        self._handlers: dict[str, HandlerFunc] = {}
        self._default_handler = default_handler
        self._auto_answer = auto_answer

        # Middleware hooks
        self._before_handlers: list[HandlerFunc] = []
        self._after_handlers: list[HandlerFunc] = []
        self._error_handlers: list[
            Callable[[Update, ContextTypes.DEFAULT_TYPE, Exception], Awaitable[None]]
        ] = []

    def handler(self, name: str) -> Callable[[HandlerFunc], HandlerFunc]:
        """Decorator to register a handler function.

        Args:
            name: Handler name (must match handler in MenuAction)

        Returns:
            Decorator function

        Example:
            >>> @router.handler("edit_user")
            >>> async def handle_edit(update, context, params):
            ...     user_id = params["user_id"]
            ...     await update.callback_query.edit_message_text(f"Editing user {user_id}")
        """

        def decorator(func: HandlerFunc) -> HandlerFunc:
            self.register_handler(name, func)
            return func

        return decorator

    def register_handler(self, name: str, func: HandlerFunc) -> None:
        """Register a handler function.

        Args:
            name: Handler name
            func: Async handler function

        Example:
            >>> async def my_handler(update, context, params):
            ...     pass
            >>> router.register_handler("my_action", my_handler)
        """
        if name in self._handlers:
            logger.warning(f"Overwriting existing handler for '{name}'")

        self._handlers[name] = func

    def register_handlers(self, handlers: Mapping[str, HandlerFunc]) -> None:
        """Register multiple handlers at once.

        Args:
            handlers: Mapping of handler names to functions

        Example:
            >>> router.register_handlers({
            ...     "action1": handle_action1,
            ...     "action2": handle_action2,
            ... })
        """
        for name, func in handlers.items():
            self.register_handler(name, func)

    def unregister_handler(self, name: str) -> bool:
        """Unregister a handler.

        Args:
            name: Handler name to remove

        Returns:
            True if handler was removed, False if not found
        """
        if name in self._handlers:
            del self._handlers[name]
            return True
        return False

    def set_default_handler(self, func: HandlerFunc) -> None:
        """Set the default fallback handler.

        Args:
            func: Async handler function
        """
        self._default_handler = func

    async def route(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """Route a callback query to the appropriate handler.

        This is the main entry point that should be registered with
        CallbackQueryHandler in python-telegram-bot.

        Args:
            update: Telegram Update object
            context: Telegram Context object

        Example:
            >>> app.add_handler(CallbackQueryHandler(router.route))
        """
        if not update.callback_query:
            logger.warning("route() called with Update without callback_query")
            return

        callback_query = update.callback_query
        callback_data = callback_query.data

        if not callback_data:
            logger.warning("Callback query has no data")
            if self._auto_answer:
                await callback_query.answer()
            return

        try:
            action = await self._encoder.decode(callback_data)
            handled = await self._dispatch(update, context, callback_query, action)
            if handled and self._auto_answer:
                await callback_query.answer()
        except DecodingError as e:
            logger.error(f"Failed to decode callback data: {e}")
            await self._handle_routing_error(
                update, context, callback_query, e, "Invalid or expired action"
            )
        except Exception as e:
            logger.exception(f"Error handling callback query: {e}")
            await self._handle_routing_error(
                update, context, callback_query, e, "An error occurred"
            )

    async def _dispatch(
        self,
        update: Update,
        context: ContextTypes.DEFAULT_TYPE,
        callback_query: CallbackQuery,
        action: MenuAction,
    ) -> bool:
        """Run middleware and execute the handler for a decoded action.

        Args:
            update: Telegram Update object.
            context: Telegram Context object.
            callback_query: The update's callback query (already validated).
            action: The decoded MenuAction.

        Returns:
            True if a handler (or the default handler) ran, False if no handler
            was found and no default handler is configured.
        """
        params = action.params
        handler_name = action.handler

        logger.debug(f"Routing to handler '{handler_name}' with params: {params}")

        # Run before middleware
        for before_handler in self._before_handlers:
            await before_handler(update, context, params)

        # Find and execute handler
        handler = self._handlers.get(handler_name)

        if handler:
            await handler(update, context, params)
        elif self._default_handler:
            logger.debug(f"Handler '{handler_name}' not found, using default handler")
            await self._default_handler(update, context, params)
        else:
            logger.warning(f"No handler registered for '{handler_name}' and no default handler")
            if self._auto_answer:
                await callback_query.answer("Action not available")
            return False

        # Run after middleware
        for after_handler in self._after_handlers:
            await after_handler(update, context, params)

        return True

    async def _handle_routing_error(
        self,
        update: Update,
        context: ContextTypes.DEFAULT_TYPE,
        callback_query: CallbackQuery,
        error: Exception,
        answer_text: str,
    ) -> None:
        """Run registered error handlers and optionally answer the callback query.

        Args:
            update: Telegram Update object.
            context: Telegram Context object.
            callback_query: The update's callback query (already validated).
            error: The exception that occurred during routing.
            answer_text: Text to show the user when auto-answering.
        """
        for error_handler in self._error_handlers:
            await error_handler(update, context, error)

        if self._auto_answer:
            await callback_query.answer(answer_text)

    def before(self, func: HandlerFunc) -> HandlerFunc:
        """Register a before middleware handler.

        These run before the main handler for every callback.

        Args:
            func: Async middleware function

        Returns:
            The function (for use as decorator)

        Example:
            >>> @router.before
            >>> async def log_callback(update, context, params):
            ...     logger.info(f"Callback received: {params}")
        """
        self._before_handlers.append(func)
        return func

    def after(self, func: HandlerFunc) -> HandlerFunc:
        """Register an after middleware handler.

        These run after the main handler for every callback.

        Args:
            func: Async middleware function

        Returns:
            The function (for use as decorator)

        Example:
            >>> @router.after
            >>> async def cleanup(update, context, params):
            ...     # Cleanup logic
            ...     pass
        """
        self._after_handlers.append(func)
        return func

    def on_error(
        self, func: Callable[[Update, ContextTypes.DEFAULT_TYPE, Exception], Awaitable[None]]
    ) -> Callable[[Update, ContextTypes.DEFAULT_TYPE, Exception], Awaitable[None]]:
        """Register an error handler.

        These run when an exception occurs during routing or handling.

        Args:
            func: Async error handler function

        Returns:
            The function (for use as decorator)

        Example:
            >>> @router.on_error
            >>> async def handle_error(update, context, error):
            ...     logger.error(f"Error: {error}")
            ...     await update.callback_query.answer("Something went wrong")
        """
        self._error_handlers.append(func)
        return func

    def get_handler(self, name: str) -> HandlerFunc | None:
        """Get a registered handler by name.

        Args:
            name: Handler name

        Returns:
            Handler function or None if not found
        """
        return self._handlers.get(name)

    def list_handlers(self) -> list[str]:
        """Get list of all registered handler names.

        Returns:
            List of handler names
        """
        return list(self._handlers.keys())

    async def claim(self, key: str, user_id: int, *, ttl: int | None = 3600) -> bool:
        """Atomically claim a key for a single user (single-winner lock).

        Backed by :meth:`~telegram_menu_builder.storage.base.StorageBackend.add`,
        which is an atomic set-if-absent: when several users race to claim the
        same key, exactly one wins. An expired claim (past its ``ttl``) is
        reclaimable, so the next caller after expiry can win the key again.

        The stored record is ``{"user_id": user_id, "claimed_at": <UTC ISO8601>}``
        and can be read back via :meth:`who_claimed`.

        Args:
            key: The resource identifier being claimed (e.g. ``"doc:1"``).
            user_id: The Telegram user id attempting the claim.
            ttl: Time-to-live for the claim in seconds (``None`` = no expiry).
                Defaults to one hour.

        Returns:
            ``True`` if this user won the claim, ``False`` if a live claim already
            existed (held by this or another user).
        """
        record = {
            "user_id": user_id,
            "claimed_at": datetime.datetime.now(datetime.UTC).isoformat(),
        }
        return await self._storage.add(key, record, ttl=ttl)

    async def who_claimed(self, key: str) -> dict[str, Any] | None:
        """Return the current claim record for ``key``, if any.

        Args:
            key: The resource identifier to inspect.

        Returns:
            The stored claim record (e.g. ``{"user_id": 7, "claimed_at": ...}``),
            or ``None`` if the key is unclaimed or the claim has expired.
        """
        return await self._storage.get(key)

    async def release(self, key: str) -> bool:
        """Release a claim, freeing the key for a future :meth:`claim`.

        Args:
            key: The resource identifier whose claim should be released.

        Returns:
            ``True`` if a claim was removed, ``False`` if the key was not claimed.
        """
        return await self._storage.delete(key)

    @property
    def storage(self) -> StorageBackend:
        """Get the storage backend."""
        return self._storage

    @property
    def encoder(self) -> CallbackEncoder:
        """Get the callback encoder."""
        return self._encoder

storage property 

storage: StorageBackend

Get the storage backend.

encoder property 

encoder: CallbackEncoder

Get the callback encoder.

handler 

handler(name: str) -> Callable[[HandlerFunc], HandlerFunc]

Decorator to register a handler function.

Parameters:

Name Type Description Default
name str

Handler name (must match handler in MenuAction)

 required

Returns:

Type Description
Callable[[HandlerFunc], HandlerFunc]

Decorator function
Example

@router.handler("edit_user") async def handle_edit(update, context, params): ... user_id = params["user_id"] ... await update.callback_query.edit_message_text(f"Editing user {user_id}")

Source code in src/telegram_menu_builder/router.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def handler(self, name: str) -> Callable[[HandlerFunc], HandlerFunc]:
    """Decorator to register a handler function.

    Args:
        name: Handler name (must match handler in MenuAction)

    Returns:
        Decorator function

    Example:
        >>> @router.handler("edit_user")
        >>> async def handle_edit(update, context, params):
        ...     user_id = params["user_id"]
        ...     await update.callback_query.edit_message_text(f"Editing user {user_id}")
    """

    def decorator(func: HandlerFunc) -> HandlerFunc:
        self.register_handler(name, func)
        return func

    return decorator

register_handler 

register_handler(name: str, func: HandlerFunc) -> None

Register a handler function.

Parameters:

Name Type Description Default
name str

Handler name

 required
func HandlerFunc

Async handler function

 required
Example

async def my_handler(update, context, params): ... pass router.register_handler("my_action", my_handler)

Source code in src/telegram_menu_builder/router.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def register_handler(self, name: str, func: HandlerFunc) -> None:
    """Register a handler function.

    Args:
        name: Handler name
        func: Async handler function

    Example:
        >>> async def my_handler(update, context, params):
        ...     pass
        >>> router.register_handler("my_action", my_handler)
    """
    if name in self._handlers:
        logger.warning(f"Overwriting existing handler for '{name}'")

    self._handlers[name] = func

register_handlers 

register_handlers(handlers: Mapping[str, HandlerFunc]) -> None

Register multiple handlers at once.

Parameters:

Name Type Description Default
handlers Mapping[str, HandlerFunc]

Mapping of handler names to functions

 required
Example

router.register_handlers({ ... "action1": handle_action1, ... "action2": handle_action2, ... })

Source code in src/telegram_menu_builder/router.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def register_handlers(self, handlers: Mapping[str, HandlerFunc]) -> None:
    """Register multiple handlers at once.

    Args:
        handlers: Mapping of handler names to functions

    Example:
        >>> router.register_handlers({
        ...     "action1": handle_action1,
        ...     "action2": handle_action2,
        ... })
    """
    for name, func in handlers.items():
        self.register_handler(name, func)

unregister_handler 

unregister_handler(name: str) -> bool

Unregister a handler.

Parameters:

Name Type Description Default
name str

Handler name to remove

 required

Returns:

Type Description
bool

True if handler was removed, False if not found
Source code in src/telegram_menu_builder/router.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
def unregister_handler(self, name: str) -> bool:
    """Unregister a handler.

    Args:
        name: Handler name to remove

    Returns:
        True if handler was removed, False if not found
    """
    if name in self._handlers:
        del self._handlers[name]
        return True
    return False

set_default_handler 

set_default_handler(func: HandlerFunc) -> None

Set the default fallback handler.

Parameters:

Name Type Description Default
func HandlerFunc

Async handler function

 required
Source code in src/telegram_menu_builder/router.py 
144
145
146
147
148
149
150
def set_default_handler(self, func: HandlerFunc) -> None:
    """Set the default fallback handler.

    Args:
        func: Async handler function
    """
    self._default_handler = func

route async 

route(update: Update, context: DEFAULT_TYPE) -> None

Route a callback query to the appropriate handler.

This is the main entry point that should be registered with CallbackQueryHandler in python-telegram-bot.

Parameters:

Name Type Description Default
update Update

Telegram Update object

 required
context DEFAULT_TYPE

Telegram Context object

 required
Example

app.add_handler(CallbackQueryHandler(router.route))

Source code in src/telegram_menu_builder/router.py 
152
153
154
155
156
157
158
159
160
161
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
async def route(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Route a callback query to the appropriate handler.

    This is the main entry point that should be registered with
    CallbackQueryHandler in python-telegram-bot.

    Args:
        update: Telegram Update object
        context: Telegram Context object

    Example:
        >>> app.add_handler(CallbackQueryHandler(router.route))
    """
    if not update.callback_query:
        logger.warning("route() called with Update without callback_query")
        return

    callback_query = update.callback_query
    callback_data = callback_query.data

    if not callback_data:
        logger.warning("Callback query has no data")
        if self._auto_answer:
            await callback_query.answer()
        return

    try:
        action = await self._encoder.decode(callback_data)
        handled = await self._dispatch(update, context, callback_query, action)
        if handled and self._auto_answer:
            await callback_query.answer()
    except DecodingError as e:
        logger.error(f"Failed to decode callback data: {e}")
        await self._handle_routing_error(
            update, context, callback_query, e, "Invalid or expired action"
        )
    except Exception as e:
        logger.exception(f"Error handling callback query: {e}")
        await self._handle_routing_error(
            update, context, callback_query, e, "An error occurred"
        )

before 

before(func: HandlerFunc) -> HandlerFunc

Register a before middleware handler.

These run before the main handler for every callback.

Parameters:

Name Type Description Default
func HandlerFunc

Async middleware function

 required

Returns:

Type Description
HandlerFunc

The function (for use as decorator)
Example

@router.before async def log_callback(update, context, params): ... logger.info(f"Callback received: {params}")

Source code in src/telegram_menu_builder/router.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def before(self, func: HandlerFunc) -> HandlerFunc:
    """Register a before middleware handler.

    These run before the main handler for every callback.

    Args:
        func: Async middleware function

    Returns:
        The function (for use as decorator)

    Example:
        >>> @router.before
        >>> async def log_callback(update, context, params):
        ...     logger.info(f"Callback received: {params}")
    """
    self._before_handlers.append(func)
    return func

after 

after(func: HandlerFunc) -> HandlerFunc

Register an after middleware handler.

These run after the main handler for every callback.

Parameters:

Name Type Description Default
func HandlerFunc

Async middleware function

 required

Returns:

Type Description
HandlerFunc

The function (for use as decorator)
Example

@router.after async def cleanup(update, context, params): ... # Cleanup logic ... pass

Source code in src/telegram_menu_builder/router.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def after(self, func: HandlerFunc) -> HandlerFunc:
    """Register an after middleware handler.

    These run after the main handler for every callback.

    Args:
        func: Async middleware function

    Returns:
        The function (for use as decorator)

    Example:
        >>> @router.after
        >>> async def cleanup(update, context, params):
        ...     # Cleanup logic
        ...     pass
    """
    self._after_handlers.append(func)
    return func

on_error 

on_error(func: Callable[[Update, DEFAULT_TYPE, Exception], Awaitable[None]]) -> Callable[[Update, ContextTypes.DEFAULT_TYPE, Exception], Awaitable[None]]

Register an error handler.

These run when an exception occurs during routing or handling.

Parameters:

Name Type Description Default
func Callable[[Update, DEFAULT_TYPE, Exception], Awaitable[None]]

Async error handler function

 required

Returns:

Type Description
Callable[[Update, DEFAULT_TYPE, Exception], Awaitable[None]]

The function (for use as decorator)
Example

@router.on_error async def handle_error(update, context, error): ... logger.error(f"Error: {error}") ... await update.callback_query.answer("Something went wrong")

Source code in src/telegram_menu_builder/router.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def on_error(
    self, func: Callable[[Update, ContextTypes.DEFAULT_TYPE, Exception], Awaitable[None]]
) -> Callable[[Update, ContextTypes.DEFAULT_TYPE, Exception], Awaitable[None]]:
    """Register an error handler.

    These run when an exception occurs during routing or handling.

    Args:
        func: Async error handler function

    Returns:
        The function (for use as decorator)

    Example:
        >>> @router.on_error
        >>> async def handle_error(update, context, error):
        ...     logger.error(f"Error: {error}")
        ...     await update.callback_query.answer("Something went wrong")
    """
    self._error_handlers.append(func)
    return func

get_handler 

get_handler(name: str) -> HandlerFunc | None

Get a registered handler by name.

Parameters:

Name Type Description Default
name str

Handler name

 required

Returns:

Type Description
HandlerFunc | None

Handler function or None if not found
Source code in src/telegram_menu_builder/router.py 
326
327
328
329
330
331
332
333
334
335
def get_handler(self, name: str) -> HandlerFunc | None:
    """Get a registered handler by name.

    Args:
        name: Handler name

    Returns:
        Handler function or None if not found
    """
    return self._handlers.get(name)

list_handlers 

list_handlers() -> list[str]

Get list of all registered handler names.

Returns:

Type Description
list[str]

List of handler names
Source code in src/telegram_menu_builder/router.py 
337
338
339
340
341
342
343
def list_handlers(self) -> list[str]:
    """Get list of all registered handler names.

    Returns:
        List of handler names
    """
    return list(self._handlers.keys())

claim async 

claim(key: str, user_id: int, *, ttl: int | None = 3600) -> bool

Atomically claim a key for a single user (single-winner lock).

Backed by :meth:~telegram_menu_builder.storage.base.StorageBackend.add, which is an atomic set-if-absent: when several users race to claim the same key, exactly one wins. An expired claim (past its ttl) is reclaimable, so the next caller after expiry can win the key again.

The stored record is {"user_id": user_id, "claimed_at": <UTC ISO8601>} and can be read back via :meth:who_claimed.

Parameters:

Name Type Description Default
key str

The resource identifier being claimed (e.g. "doc:1").

 required
user_id int

The Telegram user id attempting the claim.

 required
ttl int | None

Time-to-live for the claim in seconds (None = no expiry). Defaults to one hour.

 3600

Returns:

Type Description
bool

True if this user won the claim, False if a live claim already
bool

existed (held by this or another user).
Source code in src/telegram_menu_builder/router.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
async def claim(self, key: str, user_id: int, *, ttl: int | None = 3600) -> bool:
    """Atomically claim a key for a single user (single-winner lock).

    Backed by :meth:`~telegram_menu_builder.storage.base.StorageBackend.add`,
    which is an atomic set-if-absent: when several users race to claim the
    same key, exactly one wins. An expired claim (past its ``ttl``) is
    reclaimable, so the next caller after expiry can win the key again.

    The stored record is ``{"user_id": user_id, "claimed_at": <UTC ISO8601>}``
    and can be read back via :meth:`who_claimed`.

    Args:
        key: The resource identifier being claimed (e.g. ``"doc:1"``).
        user_id: The Telegram user id attempting the claim.
        ttl: Time-to-live for the claim in seconds (``None`` = no expiry).
            Defaults to one hour.

    Returns:
        ``True`` if this user won the claim, ``False`` if a live claim already
        existed (held by this or another user).
    """
    record = {
        "user_id": user_id,
        "claimed_at": datetime.datetime.now(datetime.UTC).isoformat(),
    }
    return await self._storage.add(key, record, ttl=ttl)

who_claimed async 

who_claimed(key: str) -> dict[str, Any] | None

Return the current claim record for key, if any.

Parameters:

Name Type Description Default
key str

The resource identifier to inspect.

 required

Returns:

Type Description
dict[str, Any] | None

The stored claim record (e.g. {"user_id": 7, "claimed_at": ...}),
dict[str, Any] | None

or None if the key is unclaimed or the claim has expired.
Source code in src/telegram_menu_builder/router.py 
372
373
374
375
376
377
378
379
380
381
382
async def who_claimed(self, key: str) -> dict[str, Any] | None:
    """Return the current claim record for ``key``, if any.

    Args:
        key: The resource identifier to inspect.

    Returns:
        The stored claim record (e.g. ``{"user_id": 7, "claimed_at": ...}``),
        or ``None`` if the key is unclaimed or the claim has expired.
    """
    return await self._storage.get(key)

release async 

release(key: str) -> bool

Release a claim, freeing the key for a future :meth:claim.

Parameters:

Name Type Description Default
key str

The resource identifier whose claim should be released.

 required

Returns:

Type Description
bool

True if a claim was removed, False if the key was not claimed.
Source code in src/telegram_menu_builder/router.py 
384
385
386
387
388
389
390
391
392
393
async def release(self, key: str) -> bool:
    """Release a claim, freeing the key for a future :meth:`claim`.

    Args:
        key: The resource identifier whose claim should be released.

    Returns:
        ``True`` if a claim was removed, ``False`` if the key was not claimed.
    """
    return await self._storage.delete(key)

RouterGroup

telegram_menu_builder.router.RouterGroup

Group multiple routers with a common prefix.

This is useful for organizing handlers by feature or module.

Example

users = RouterGroup("users", main_router) @users.handler("edit") async def edit_user(update, context, params): ... # Handler will be registered as "users.edit" ... pass

Source code in src/telegram_menu_builder/router.py 
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
class RouterGroup:
    """Group multiple routers with a common prefix.

    This is useful for organizing handlers by feature or module.

    Example:
        >>> users = RouterGroup("users", main_router)
        >>> @users.handler("edit")
        >>> async def edit_user(update, context, params):
        ...     # Handler will be registered as "users.edit"
        ...     pass
    """

    def __init__(self, prefix: str, router: MenuRouter) -> None:
        """Initialize router group.

        Args:
            prefix: Prefix for all handlers in this group
            router: Parent MenuRouter instance
        """
        self.prefix = prefix
        self.router = router

    def handler(self, name: str) -> Callable[[HandlerFunc], HandlerFunc]:
        """Decorator to register a handler with prefix.

        Args:
            name: Handler name (will be prefixed)

        Returns:
            Decorator function
        """
        full_name = f"{self.prefix}.{name}"
        return self.router.handler(full_name)

    def register_handler(self, name: str, func: HandlerFunc) -> None:
        """Register a handler with prefix.

        Args:
            name: Handler name (will be prefixed)
            func: Async handler function
        """
        full_name = f"{self.prefix}.{name}"
        self.router.register_handler(full_name, func)

handler 

handler(name: str) -> Callable[[HandlerFunc], HandlerFunc]

Decorator to register a handler with prefix.

Parameters:

Name Type Description Default
name str

Handler name (will be prefixed)

 required

Returns:

Type Description
Callable[[HandlerFunc], HandlerFunc]

Decorator function
Source code in src/telegram_menu_builder/router.py 
429
430
431
432
433
434
435
436
437
438
439
def handler(self, name: str) -> Callable[[HandlerFunc], HandlerFunc]:
    """Decorator to register a handler with prefix.

    Args:
        name: Handler name (will be prefixed)

    Returns:
        Decorator function
    """
    full_name = f"{self.prefix}.{name}"
    return self.router.handler(full_name)

register_handler 

register_handler(name: str, func: HandlerFunc) -> None

Register a handler with prefix.

Parameters:

Name Type Description Default
name str

Handler name (will be prefixed)

 required
func HandlerFunc

Async handler function

 required
Source code in src/telegram_menu_builder/router.py 
441
442
443
444
445
446
447
448
449
def register_handler(self, name: str, func: HandlerFunc) -> None:
    """Register a handler with prefix.

    Args:
        name: Handler name (will be prefixed)
        func: Async handler function
    """
    full_name = f"{self.prefix}.{name}"
    self.router.register_handler(full_name, func)