Repository: apache/airflow
File: providers/common/sql/src/airflow/providers/common/sql/decorators/sql.py (Python)

Code before:

    @property
    def xcom_push(self) -> bool:
        """Compatibility property for BaseDecorator that expects xcom_push attribute."""
        return self.do_xcom_push

    @xcom_push.setter
    def xcom_push(self, value: bool) -> None:
        """Compatibility setter for BaseDecorator that expects xcom_push attribute."""
        self.do_xcom_push = value

    def execute(self, context: Context) -> Any:
        """
        Build the SQL and execute the generated query (or queries).

        :param context: Airflow context.
        :return: Any
        """
        context_merge(context, self.op_kwargs)
        kwargs = determine_kwargs(self.python_callable, self.op_args, context)

        # Set the sql
        self.sql = self.python_callable(*self.op_args, **kwargs)

        if not isinstance(self.sql, str) or self.sql.strip() == "":
            raise TypeError("The returned value from the TaskFlow callable must be a non-empty string.")

        self.render_template_fields(context)

        return super().execute(context)


def sql_task(python_callable=None, **kwargs) -> TaskDecorator:
    """
    Wrap a Python function into a SQLExecuteQueryOperator.

    :param python_callable: Function to decorate.

    :meta private:
    """
    return task_decorator_factory(
        python_callable=python_callable,
        decorated_operator_class=_SQLDecoratedOperator,
        **kwargs,
    )


Review comment (refactor):
```suggestion
        if not self.sql or (
    isinstance(self.sql, str) and not self.sql.strip()
) or (
    isinstance(self.sql, list)
    and not all(isinstance(s, str) and s.strip() for s in self.sql)
) or not isinstance(self.sql, (str, list)):
    raise TypeError(
        "The returned value from the TaskFlow callable must be a non-empty string "
        "or a non-empty list of non-empty strings."
    )
```
I think `SqlExecuteQueryOperator` allows list[str] as well, so if we do check return type from python, we might as well support it here. Other option is to remove this type check entirely and let the Sql Operator do the type check for us and raise? In case it evolves in the future, I think this might be the best choice.

Code after:
        )

    @property
    def xcom_push(self) -> bool:
        """Compatibility property for BaseDecorator that expects xcom_push attribute."""
        return self.do_xcom_push

    @xcom_push.setter
    def xcom_push(self, value: bool) -> None:
        """Compatibility setter for BaseDecorator that expects xcom_push attribute."""
        self.do_xcom_push = value

    def execute(self, context: Context) -> Any:
        """
        Build the SQL and execute the generated query (or queries).

        :param context: Airflow context.
        :return: Any
        """
        context_merge(context, self.op_kwargs)
        kwargs = determine_kwargs(self.python_callable, self.op_args, context)

        # Set the sql using the Python callable
        self.sql = self.python_callable(*self.op_args, **kwargs)

        # Only non-empty strings and lists of non-empty strings are acceptable return types
        if (
            not isinstance(self.sql, (str, list))
            or (isinstance(self.sql, str) and not self.sql.strip())
            or (isinstance(self.sql, list) and not all(isinstance(s, str) and s.strip() for s in self.sql))
        ):
            raise TypeError(
                "The returned value from the TaskFlow callable must be a non-empty "
                "string or a non-empty list of non-empty strings."
            )

        self.render_template_fields(context)

        return super().execute(context)


def sql_task(python_callable=None, **kwargs) -> TaskDecorator:
    """
    Wrap a Python function into a SQLExecuteQueryOperator.

    :param python_callable: Function to decorate.

    :meta private:
    """
    return task_decorator_factory(
        python_callable=python_callable,