def dapr_topic_trigger()

in azure/functions/decorators/function_app.py [0:0]

• 25 lines of code
• 5 McCabe index (conditional complexity)

    def dapr_topic_trigger(self,
                           arg_name: str,
                           pub_sub_name: str,
                           topic: str,
                           route: Optional[str] = None,
                           data_type: Optional[
                               Union[DataType, str]] = None,
                           **kwargs: Any) -> Callable[..., Any]:
        """The dapr_topic_trigger decorator adds
        :class:`DaprTopicTrigger`
        to the :class:`FunctionBuilder` object
        for building :class:`Function` object used in worker function
        indexing model. This is equivalent to defining DaprTopicTrigger
        in the function.json which enables function to be triggered when new
        message(s) are sent to the Dapr pubsub.
        All optional fields will be given default value by function host when
        they are parsed by function host.

        Ref: https://aka.ms/azure-function-dapr-trigger-topic

        :param arg_name: The name of the variable that represents
        :param pub_sub_name: The pub/sub name.
        :param topic: The topic. If unspecified the function name will be used.
        :param route: The route for the trigger. If unspecified
        the topic name will be used.
        :param data_type: Defines how Functions runtime should treat the
        parameter value.
        :param kwargs: Keyword arguments for specifying additional binding
        fields to include in the binding json.

        :return: Decorator function.
        """
        # TODO: This is a temporary check, it should be removed once route
        # issue is fixed at python worker.
        # Currently, python worker treats route as HttpTrigger attribute and
        # expects value for route. Route could be nil for dapr topic trigger.
        if not route:
            route = topic

        @self._configure_function_builder
        def wrap(fb):
            def decorator():
                fb.add_trigger(
                    trigger=DaprTopicTrigger(
                        name=arg_name,
                        pub_sub_name=pub_sub_name,
                        topic=topic,
                        route=route,
                        data_type=parse_singular_param_to_enum(data_type,
                                                               DataType),
                        **kwargs))
                return fb

            return decorator()

        return wrap