Skip to content

kamihi.tg.handlers.auth_handler ⚓︎

Custom handler for Telegram bot that checks if a user is authorized to use a wrapped handler.

License

MIT

Classes:

Name Description
AuthHandler

Custom wrapper handler that checks if the user is authorized to use the wrapped handler before executing it.

AuthHandler ⚓︎

AuthHandler(handler: BaseHandler, name: str)

Bases: BaseHandler

Custom wrapper handler that checks if the user is authorized to use the wrapped handler before executing it.

Attributes:

Name Type Description
handler BaseHandler

the handler to be wrapped.

name str

The name of the action.

Initialize the AuthHandler with the callback function.

Methods:

Name Description
check_update

Determine if an update should be handled by this handler instance.

Source code in src/kamihi/tg/handlers/auth_handler.py
29
30
31
32
33
def __init__(self, handler: BaseHandler, name: str) -> None:
    """Initialize the AuthHandler with the callback function."""
    self.handler = handler
    self.name = name
    super().__init__(self.handler.callback)

check_update ⚓︎

check_update(update: Update) -> bool

Determine if an update should be handled by this handler instance.

Source code in src/kamihi/tg/handlers/auth_handler.py
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def check_update(self, update: Update) -> bool:
    """Determine if an update should be handled by this handler instance."""
    if not isinstance(update, Update):
        return False

    if update.message and update.effective_user:
        user = get_user_from_telegram_id(update.effective_user.id)

        if user is None:
            logger.bind(user_id=update.effective_user.id, action=self.name).debug(
                "User not found in the database tried to use action."
            )
            return False

        if not is_user_authorized(user, self.name):
            logger.bind(user_id=user.telegram_id, action=self.name).debug(
                "User is not authorized to use this action."
            )
            return False

    return self.handler.check_update(update)