Function async_send_notification

src/sentry/notifications/utils/tasks.py:71–106

This function takes a notification class and arguments to instantiate the notification class in a task with the original arguments. It converts instances of models into a JSON object that allows us to query the rows inside the task so the interface is just simple JSON.

(
    NotificationClass: type[BaseNotification], *args: Any, **kwargs: Any
)

@cell_silo_function
def async_send_notification(
    NotificationClass: type[BaseNotification], *args: Any, **kwargs: Any
) -> None:
    """
    This function takes a notification class and arguments to instantiate
    the notification class in a task with the original arguments.
    It converts instances of models into a JSON object that allows us to query
    the rows inside the task so the interface is just simple JSON.
    """
    class_name = getattr(NotificationClass, "__name__")
    if not get(class_name):
        raise NotificationClassNotSetException()

    task_args = []
    for arg in args:
        if isinstance(arg, Model):
            task_args.append(serialize_model(arg))
        elif isinstance(arg, SimpleLazyObject):
            task_args.append(serialize_lazy_object_user(arg))
        elif isinstance(arg, AnonymousUser):
            task_args.append(serialize_anonymous_user(arg))
        # maybe we need an explicit check if it's a primitive?
        else:
            task_args.append({"type": "other", "value": arg, "key": None})
    for key, val in kwargs.items():
        if isinstance(val, Model):
            task_args.append(serialize_model(val, key))
        elif isinstance(val, SimpleLazyObject):
            task_args.append(serialize_lazy_object_user(val, key))
        elif isinstance(val, AnonymousUser):
            task_args.append(serialize_anonymous_user(val, key))
        # maybe we need an explicit check if it's a primitive?
        else:
            task_args.append({"type": "other", "value": val, "key": key})

    _send_notification.delay(class_name, task_args)


@instrumented_task(