Using Python decorators to process and authorize requests

– One syntactic sugar pill to rule them all.

When building any system that interacts with users you always need to check that they are who they are claiming to be and whether they are allowed to do what they are trying to do. In other words, you need to authenticate and authorize users.

Chatbots are no exception. Strictly speaking, with chatbots you do not need to authenticate users yourself – the platform does it for you. However, on each request you still have to load the user model and authorize it.

Below I will show how I approached authorization and preprocessing requests in my Telegram bot for shopping and to-do lists listOK (project page to avoid code duplication and make it more explicit. This approach is not limited to Telegram or chatbots in general. It can be applied in any request-centered projects, which in our API date and age are the majority.

The code below is close to what I use with some simplifications and omissions for brevity.

Naive approach

To communicate with Telegram bot-API I use the python-telegram-bot library. It processes each message or other type of communication via separate handlers: functions that receive update details and overall bot context.

In listOK each user has lists and each list consists of items. This means that all handlers for creating, reading, updating, deleting lists and items should be able to do similar things: load required models from the database and authorize the user.

At the very beginning of the project I tried a direct approach – loading user models and authorizing them in the request handlers themselves:

def command_my_lists(
    update: Update, context: CallbackContext
) -> None:
    user = find_user_by_telegram_id(
        db.session, update.effective_user.id
    )
    if user is None:
        return
    # Display user's lists

However, very soon it became unviable: there are dozens of handlers, and repeating the same code everywhere was not DRY at all. What if I wanted to change the logic or react to an unauthorized user request?

Loading models with decorators

Here Python decorators came in very handy. They allow adding behavior and pre-processing to any function without modifying it:

def load_user_or_fail(fn):
    """Decorator: loads User model and passes it to the function
    or stops the request."""
    def wrapper(*args, **kwargs):
        # Expects that Update object is always the first arg
        update: Update = args[0]
        user = find_user_by_telegram_id(
            db.session, update.effective_user.id
        )

        # Ignore requests from unknown users
        if user is None:
            return
        return fn(*args, **kwargs, user=user)

    return wrapper

Now I can add user loading to any handler with only one line and additional argument in the handler.

@load_user_or_fail
def command_my_lists(
    update: Update,
    context: CallbackContext,
    user: User
) -> None:
    # Display user's lists

I use this decorator for all handlers except for the /start command, since new users need it to register. It handles the user registration/loading itself.

The same is true for loading other models. For instance, all handlers that work with lists need to load the respective models. It also can be moved to a decorator to reduce the boilerplate code:

def load_list_or_fail(source):
    def load_list_or_fail_outer_wrapper(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # Expects the update and context objects 
            # to be first arguments
            update: Update = args[0]
            context: CallbackContext = args[1]
            list_id = None

            if source == "callback_data":
                list_id = int(re.findall(
                    r"[0-9]+", update.callback_query.data
                )[0])

            if source == "user_data":
                list_id = context.user_data.get(
                    "active_list_id", None
                )

            items_list = find_list_by_id(db.session, list_id)

            # Ignore request for non-existing list
            if items_list is None:
                logging.getLogger().error(
                    "Could not load list."
                )
                return

            return fn(*args, **kwargs, items_list=items_list)
        return wrapper
    return load_list_or_fail_outer_wrapper

This is a bit more involved than loading a user. A list_id of the list to load can arrive from either of two sources: a callback data (string returned after a user presses an inline keyboard button) or stored in user context (required for multi-step communications). To accommodate this I added a source argument to the decorator.

It adds complexity: I need to provide in advance what source to use. I thought about making source detection automatic but decided against it. First, sometimes both sources can be legitimately present. Second, this would have introduced unnecessary at that moment 'magic' and reduced code readability.

Authorize with decorators

Decorators help with authorization as well, although their implementation will vary greatly from project to project. For example, here is what I did to check permissions for updating a list name:

from enum import Enum, auto


class Permissions(Enum):
    """Permissions constants"""
    CREATE = auto()
    READ = auto()
    UPDATE = auto()
    DELETE = auto()


def permission_required(permission):
    """Decorator for checking permissions"""
    def outer_wrapper(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            is_authorized = False
            keys = kwargs.keys()

            if "items_list" in keys:
                target = kwargs["items_list"]
            elif "list_item" in keys:
                target = kwargs["list_item"]

            is_authorized = target.check_user_permission(
                kwargs["user"], permission
            )
            if not is_authorized:
                return

            return fn(*args, **kwargs)
        return wrapper
    return outer_wrapper


@load_user_or_fail
@load_list_or_fail(source="user_data")
@permission_required(Permissions.UPDATE)
def message_rename_list(
    update: Update,
    context: CallbackContext,
    user: User,
    items_list: ItemsList
) -> int:
    # Update list name

Here I chained three decorators: two that we saw before and a new one:

1. @load_user_or_fail loads the user model and passes it as a keyword argument.

2. @load_list_or_fail(source="user_data") loads a list model based on user_data context and passes it further as a keyword argument as well.

3. Finally, @permission_required(Permissions.UPDATE) checks whether items_list keyword argument is present, and if it is, calls list's method check_user_permission which checks if the loaded user has the permission required.

If instead of a list we loaded a list item, the permission_required decorator would have detected it and called the item's check_user_permission method.

Unlike in load_list_or_fail, here I opted to automatically detect the target. It does not introduce any magic: by controlling the decorator I am calling before checking the permission I control the target.

Note: @permission_required will fail if kwargs contain nether items_list nor list_item. This is by design: it means that I forgot a loading decorator and requires immediate attention. If it happens in production, the bot will send me a message with all the exception details.

These decorators made the code much more concise and DRY, saving me a lot of time. What I especially like about this approach is how modular and explicit it is.