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

Code before:
    """

    template_fields: Sequence[str] = (
        *DecoratedOperator.template_fields,
        *SQLExecuteQueryOperator.template_fields,
    )
    template_fields_renderers: ClassVar[dict[str, str]] = {
        **DecoratedOperator.template_fields_renderers,
        **SQLExecuteQueryOperator.template_fields_renderers,
    }

    custom_operator_name: str = "@task.sql"

    def __init__(
        self,
        python_callable: Callable,
        op_args: Collection[Any] | None = None,
        op_kwargs: Mapping[str, Any] | None = None,
        **kwargs,
    ) -> None:
        super().__init__(
            python_callable=python_callable,
            op_args=op_args,
            op_kwargs=op_kwargs,
            sql=SET_DURING_EXECUTION,
            **kwargs,
        )

    @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)


Review comment (bug):
Using `SET_DURING_EXECUTION` exposed a lack of one type check in the OL method of BaseSQLOperator. Would you mind adding something like this?

```
--- a/providers/common/sql/src/airflow/providers/common/sql/operators/sql.py
+++ b/providers/common/sql/src/airflow/providers/common/sql/operators/sql.py
@@ -226,6 +226,19 @@ class BaseSQLOperator(BaseOperator):
         if not sql:
             self.log.debug("OpenLineage could not find 'sql' attribute on `%s`.", type(self).__name__)
             return OperatorLineage()
+        if (
+            not isinstance(sql, (str, list))
+            or (isinstance(sql, str) and not sql.strip())
+            or (isinstance(sql, list) and not all(isinstance(s, str) and s.strip() for s in sql))
+        ):
+            self.log.debug(
+                "OpenLineage found unsupported type of 'sql' attribute on `%s`, type=`%s`, value=`%s`. "
+                "Expected non-empty string or list of non-empty strings.",
+                type(self).__name__,
+                type(sql),
+                sql,
+            )
+            return OperatorLineage()
 
         hook = self.get_db_hook()
         try:

```

Code after:
    """

    template_fields: Sequence[str] = (
        *DecoratedOperator.template_fields,
        *SQLExecuteQueryOperator.template_fields,
    )
    template_fields_renderers: ClassVar[dict[str, str]] = {
        **DecoratedOperator.template_fields_renderers,
        **SQLExecuteQueryOperator.template_fields_renderers,
    }

    custom_operator_name: str = "@task.sql"
    overwrite_rtif_after_execution: bool = True

    def __init__(
        self,
        python_callable: Callable,
        op_args: Collection[Any] | None = None,
        op_kwargs: Mapping[str, Any] | None = None,
        **kwargs,
    ) -> None:
        super().__init__(
            python_callable=python_callable,
            op_args=op_args,
            op_kwargs=op_kwargs,
            sql=SET_DURING_EXECUTION,
            **kwargs,
        )

    @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)