You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/08/14 23:18:24 UTC
[GitHub] [airflow] kazanzhy opened a new pull request, #25717: Add SQLExecuteQueryOperator (not tested)
kazanzhy opened a new pull request, #25717:
URL: https://github.com/apache/airflow/pull/25717
There are many changes here that potentially can break backward compatibility, therefore should be tested.
Work in progress and this PR is created for discussions if some things could be implemented better.
closes: #25259
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vincbeck commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
vincbeck commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1265539311
Documentation is still missing :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
eladkal commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1303228419
We can modify the deprecated class so it will be backward compatible and add nodify the new operator to template conn_id
Would you like to raise PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kazanzhy commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
kazanzhy commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1265676602
Thanks @vincbeck, added docs.
This operator is pretty simple bu if you have something more to add feel free for suggestions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] fritz-astronomer commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
fritz-astronomer commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1281313924
I like the change - however completely removing the `execute` feels really aggressive. Shouldn't we deprecate and then remove later? This will break a *lot* of workflows if anyone's dependency get's bumped without them knowing about the change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vincbeck commented on a diff in pull request #25717: Add SQLExecuteQueryOperator (not tested)
Posted by GitBox <gi...@apache.org>.
vincbeck commented on code in PR #25717:
URL: https://github.com/apache/airflow/pull/25717#discussion_r954260289
##########
airflow/providers/apache/drill/operators/drill.py:
##########
@@ -15,16 +15,16 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-from typing import TYPE_CHECKING, Iterable, Mapping, Optional, Sequence, Union
+import warnings
+from typing import TYPE_CHECKING, Sequence
-from airflow.models import BaseOperator
-from airflow.providers.apache.drill.hooks.drill import DrillHook
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
if TYPE_CHECKING:
- from airflow.utils.context import Context
+ pass
Review Comment:
This if looks useless to me
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -63,13 +75,13 @@ class BaseSQLOperator(BaseOperator):
The provided method is .get_db_hook(). The default behavior will try to
retrieve the DB hook based on connection type.
- You can custom the behavior by overriding the .get_db_hook() method.
+ You can customize the behavior by overriding the .get_db_hook() method.
"""
def __init__(
self,
*,
- conn_id: Optional[str] = None,
+ conn_id: str,
Review Comment:
This seems not backward compatible to me
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -112,6 +124,61 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
+ """
+ Executes SQL code in a specific database
+ :param sql: the SQL code or string pointing to a template file to be executed (templated).
+ File must have a '.sql' extensions.
+ :param handler: (optional) the function that will be applied to the cursor.
+ :param parameters: (optional) the parameters to render the SQL query with.
+ :param conn_id: reference to a specific database
Review Comment:
Some parameters are missing for the docstring
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -112,6 +124,61 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
+ """
+ Executes SQL code in a specific database
+ :param sql: the SQL code or string pointing to a template file to be executed (templated).
+ File must have a '.sql' extensions.
+ :param handler: (optional) the function that will be applied to the cursor.
+ :param parameters: (optional) the parameters to render the SQL query with.
+ :param conn_id: reference to a specific database
+ """
+
+ template_fields: Sequence[str] = ('sql', 'parameters')
+ template_ext: Sequence[str] = ('.sql',)
+ ui_color = '#cdaaed'
+
+ def __init__(
+ self,
+ *,
+ sql: Union[str, List[str]],
+ autocommit: bool = False,
+ parameters: Optional[Union[Mapping, Iterable]] = None,
+ handler: Callable[[Any], Any] = fetch_all_handler,
+ split_statements: bool = False,
+ return_last: bool = True,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.sql = sql
+ self.autocommit = autocommit
+ self.parameters = parameters
+ self.handler = handler
+ self.split_statements = split_statements
+ self.return_last = return_last
+
+ def execute(self, context):
+ self.log.info('Executing: %s', self.sql)
+ hook = self.get_db_hook()
+ if self.do_xcom_push:
+ output = hook.run(
+ self.sql,
+ autocommit=self.autocommit,
+ parameters=self.parameters,
+ handler=self.handler,
+ split_statements=self.split_statements,
+ return_last=self.return_last,
+ )
+ else:
+ output = hook.run(self.sql, autocommit=self.autocommit, parameters=self.parameters)
+
+ if hasattr(self, '_process_output'):
+ for out in output:
+ self._process_output(*out)
+
+ return output
Review Comment:
After, it is up to the user to use the returned value or not (using Xcom)
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -112,6 +124,61 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
Review Comment:
Please create unit tests for this operator. You should not rely on unit tests made for all the different DB operators because at some point we are going to remove them (since they are now deprecated). Along the operators we'll remove, we'll remove as well the unit tests associated to them
##########
airflow/providers/jdbc/operators/jdbc.py:
##########
@@ -15,18 +15,16 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+import warnings
+from typing import TYPE_CHECKING, Sequence
-from typing import TYPE_CHECKING, Any, Callable, Iterable, Mapping, Optional, Sequence, Union
-
-from airflow.models import BaseOperator
-from airflow.providers.common.sql.hooks.sql import fetch_all_handler
-from airflow.providers.jdbc.hooks.jdbc import JdbcHook
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
if TYPE_CHECKING:
Review Comment:
Needed?
##########
airflow/providers/trino/operators/trino.py:
##########
@@ -16,19 +16,19 @@
# specific language governing permissions and limitations
# under the License.
"""This module contains the Trino operator."""
-
-from typing import TYPE_CHECKING, Any, Callable, List, Optional, Sequence, Union
+import warnings
+from typing import TYPE_CHECKING, Any, Sequence
from trino.exceptions import TrinoQueryError
-from airflow.models import BaseOperator
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.trino.hooks.trino import TrinoHook
if TYPE_CHECKING:
Review Comment:
Needed?
##########
airflow/providers/databricks/operators/databricks_sql.py:
##########
@@ -80,53 +80,44 @@ class DatabricksSqlOperator(BaseOperator):
def __init__(
self,
*,
- sql: Union[str, Iterable[str]],
databricks_conn_id: str = DatabricksSqlHook.default_conn_name,
http_path: Optional[str] = None,
sql_endpoint_name: Optional[str] = None,
- parameters: Optional[Union[Iterable, Mapping]] = None,
session_configuration=None,
http_headers: Optional[List[Tuple[str, str]]] = None,
catalog: Optional[str] = None,
schema: Optional[str] = None,
- do_xcom_push: bool = False,
output_path: Optional[str] = None,
output_format: str = 'csv',
csv_params: Optional[Dict[str, Any]] = None,
client_parameters: Optional[Dict[str, Any]] = None,
**kwargs,
) -> None:
- """Creates a new ``DatabricksSqlOperator``."""
- super().__init__(**kwargs)
+ super().__init__(conn_id=databricks_conn_id, **kwargs)
self.databricks_conn_id = databricks_conn_id
- self.sql = sql
- self._http_path = http_path
- self._sql_endpoint_name = sql_endpoint_name
self._output_path = output_path
self._output_format = output_format
self._csv_params = csv_params
- self.parameters = parameters
- self.do_xcom_push = do_xcom_push
- self.session_config = session_configuration
- self.http_headers = http_headers
- self.catalog = catalog
- self.schema = schema
- self.client_parameters = client_parameters or {}
- def _get_hook(self) -> DatabricksSqlHook:
- return DatabricksSqlHook(
- self.databricks_conn_id,
- http_path=self._http_path,
- session_configuration=self.session_config,
- sql_endpoint_name=self._sql_endpoint_name,
- http_headers=self.http_headers,
- catalog=self.catalog,
- schema=self.schema,
- caller="DatabricksSqlOperator",
- **self.client_parameters,
- )
+ client_parameters = {} if client_parameters is None else client_parameters
+ hook_params = kwargs.pop('hook_params', {})
- def _format_output(self, schema, results):
+ self.hook_params = {
+ 'http_path': http_path,
+ 'session_configuration': session_configuration,
+ 'sql_endpoint_name': sql_endpoint_name,
+ 'http_headers': http_headers,
+ 'catalog': catalog,
+ 'schema': schema,
+ 'caller': "DatabricksSqlOperator",
+ **client_parameters,
+ **hook_params,
+ }
+
+ def get_db_hook(self) -> DatabricksSqlHook:
+ return DatabricksSqlHook(self.databricks_conn_id, **self.hook_params)
Review Comment:
In the other operators, defining only the `hook_params` seem enough. e.g. `kwargs['hook_params'] = {'schema': schema, **hook_params}`. Why in this case you need to override `get_db_hook`?
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -112,6 +124,61 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
+ """
+ Executes SQL code in a specific database
+ :param sql: the SQL code or string pointing to a template file to be executed (templated).
+ File must have a '.sql' extensions.
+ :param handler: (optional) the function that will be applied to the cursor.
+ :param parameters: (optional) the parameters to render the SQL query with.
+ :param conn_id: reference to a specific database
+ """
+
+ template_fields: Sequence[str] = ('sql', 'parameters')
+ template_ext: Sequence[str] = ('.sql',)
+ ui_color = '#cdaaed'
+
+ def __init__(
+ self,
+ *,
+ sql: Union[str, List[str]],
+ autocommit: bool = False,
+ parameters: Optional[Union[Mapping, Iterable]] = None,
+ handler: Callable[[Any], Any] = fetch_all_handler,
+ split_statements: bool = False,
+ return_last: bool = True,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.sql = sql
+ self.autocommit = autocommit
+ self.parameters = parameters
+ self.handler = handler
+ self.split_statements = split_statements
+ self.return_last = return_last
+
+ def execute(self, context):
+ self.log.info('Executing: %s', self.sql)
+ hook = self.get_db_hook()
+ if self.do_xcom_push:
+ output = hook.run(
+ self.sql,
+ autocommit=self.autocommit,
+ parameters=self.parameters,
+ handler=self.handler,
+ split_statements=self.split_statements,
+ return_last=self.return_last,
+ )
+ else:
+ output = hook.run(self.sql, autocommit=self.autocommit, parameters=self.parameters)
+
+ if hasattr(self, '_process_output'):
+ for out in output:
+ self._process_output(*out)
+
+ return output
Review Comment:
This if is not really clear to me. Why not splitting the statements if `do_xcom_push` is `False`? How is that related? Same for `return_last`. I would do it differently:
```suggestion
output = hook.run(
self.sql,
autocommit=self.autocommit,
parameters=self.parameters,
handler=self.handler,
split_statements=self.split_statements,
return_last=self.return_last,
)
if hasattr(self, '_process_output'):
for out in output:
self._process_output(*out)
return output
```
##########
airflow/providers/postgres/operators/postgres.py:
##########
@@ -15,19 +15,19 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-from typing import TYPE_CHECKING, Iterable, Mapping, Optional, Sequence, Union
+import warnings
+from typing import TYPE_CHECKING, Mapping, Optional, Sequence
from psycopg2.sql import SQL, Identifier
-from airflow.models import BaseOperator
-from airflow.providers.postgres.hooks.postgres import PostgresHook
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.www import utils as wwwutils
if TYPE_CHECKING:
Review Comment:
Needed?
##########
airflow/providers/exasol/operators/exasol.py:
##########
@@ -15,16 +15,16 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-from typing import TYPE_CHECKING, Iterable, Mapping, Optional, Sequence, Union
+import warnings
+from typing import TYPE_CHECKING, Sequence, Optional
-from airflow.models import BaseOperator
-from airflow.providers.exasol.hooks.exasol import ExasolHook
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
if TYPE_CHECKING:
- from airflow.utils.context import Context
+ pass
Review Comment:
Is it needed?
##########
airflow/providers/vertica/operators/vertica.py:
##########
@@ -15,16 +15,16 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-from typing import TYPE_CHECKING, Any, Iterable, Sequence, Union
+import warnings
+from typing import TYPE_CHECKING, Any, Sequence
-from airflow.models import BaseOperator
-from airflow.providers.vertica.hooks.vertica import VerticaHook
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
if TYPE_CHECKING:
Review Comment:
Needed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] fritz-astronomer commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
fritz-astronomer commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1281316380
Ah that makes a lot more sense! Apologies! Thank you for the clarification, I see that now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kazanzhy commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
kazanzhy commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1259585767
@vincbeck
Sorry about that. It was many changes after my commit so I'm fixing rebase issues
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal merged pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
eladkal merged PR #25717:
URL: https://github.com/apache/airflow/pull/25717
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kazanzhy commented on a diff in pull request #25717: Add SQLExecuteQueryOperator (not tested)
Posted by GitBox <gi...@apache.org>.
kazanzhy commented on code in PR #25717:
URL: https://github.com/apache/airflow/pull/25717#discussion_r956570972
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -112,6 +124,61 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
+ """
+ Executes SQL code in a specific database
+ :param sql: the SQL code or string pointing to a template file to be executed (templated).
+ File must have a '.sql' extensions.
+ :param handler: (optional) the function that will be applied to the cursor.
+ :param parameters: (optional) the parameters to render the SQL query with.
+ :param conn_id: reference to a specific database
+ """
+
+ template_fields: Sequence[str] = ('sql', 'parameters')
+ template_ext: Sequence[str] = ('.sql',)
+ ui_color = '#cdaaed'
+
+ def __init__(
+ self,
+ *,
+ sql: Union[str, List[str]],
+ autocommit: bool = False,
+ parameters: Optional[Union[Mapping, Iterable]] = None,
+ handler: Callable[[Any], Any] = fetch_all_handler,
+ split_statements: bool = False,
+ return_last: bool = True,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.sql = sql
+ self.autocommit = autocommit
+ self.parameters = parameters
+ self.handler = handler
+ self.split_statements = split_statements
+ self.return_last = return_last
+
+ def execute(self, context):
+ self.log.info('Executing: %s', self.sql)
+ hook = self.get_db_hook()
+ if self.do_xcom_push:
+ output = hook.run(
+ self.sql,
+ autocommit=self.autocommit,
+ parameters=self.parameters,
+ handler=self.handler,
+ split_statements=self.split_statements,
+ return_last=self.return_last,
+ )
+ else:
+ output = hook.run(self.sql, autocommit=self.autocommit, parameters=self.parameters)
+
+ if hasattr(self, '_process_output'):
+ for out in output:
+ self._process_output(*out)
+
+ return output
Review Comment:
Here is one interesting thing.
If we set `None` as the default handler then results won't be returned.
Even for `do_xcom_push=True`.
In another case, if we want to keep `fetch_all_handler` as the default handler then we can't pass it for non-SELECT statements because it raises an error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kazanzhy commented on a diff in pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
kazanzhy commented on code in PR #25717:
URL: https://github.com/apache/airflow/pull/25717#discussion_r962380700
##########
airflow/providers/databricks/operators/databricks_sql.py:
##########
@@ -80,53 +80,44 @@ class DatabricksSqlOperator(BaseOperator):
def __init__(
self,
*,
- sql: Union[str, Iterable[str]],
databricks_conn_id: str = DatabricksSqlHook.default_conn_name,
http_path: Optional[str] = None,
sql_endpoint_name: Optional[str] = None,
- parameters: Optional[Union[Iterable, Mapping]] = None,
session_configuration=None,
http_headers: Optional[List[Tuple[str, str]]] = None,
catalog: Optional[str] = None,
schema: Optional[str] = None,
- do_xcom_push: bool = False,
output_path: Optional[str] = None,
output_format: str = 'csv',
csv_params: Optional[Dict[str, Any]] = None,
client_parameters: Optional[Dict[str, Any]] = None,
**kwargs,
) -> None:
- """Creates a new ``DatabricksSqlOperator``."""
- super().__init__(**kwargs)
+ super().__init__(conn_id=databricks_conn_id, **kwargs)
self.databricks_conn_id = databricks_conn_id
- self.sql = sql
- self._http_path = http_path
- self._sql_endpoint_name = sql_endpoint_name
self._output_path = output_path
self._output_format = output_format
self._csv_params = csv_params
- self.parameters = parameters
- self.do_xcom_push = do_xcom_push
- self.session_config = session_configuration
- self.http_headers = http_headers
- self.catalog = catalog
- self.schema = schema
- self.client_parameters = client_parameters or {}
- def _get_hook(self) -> DatabricksSqlHook:
- return DatabricksSqlHook(
- self.databricks_conn_id,
- http_path=self._http_path,
- session_configuration=self.session_config,
- sql_endpoint_name=self._sql_endpoint_name,
- http_headers=self.http_headers,
- catalog=self.catalog,
- schema=self.schema,
- caller="DatabricksSqlOperator",
- **self.client_parameters,
- )
+ client_parameters = {} if client_parameters is None else client_parameters
+ hook_params = kwargs.pop('hook_params', {})
- def _format_output(self, schema, results):
+ self.hook_params = {
+ 'http_path': http_path,
+ 'session_configuration': session_configuration,
+ 'sql_endpoint_name': sql_endpoint_name,
+ 'http_headers': http_headers,
+ 'catalog': catalog,
+ 'schema': schema,
+ 'caller': "DatabricksSqlOperator",
+ **client_parameters,
+ **hook_params,
+ }
+
+ def get_db_hook(self) -> DatabricksSqlHook:
+ return DatabricksSqlHook(self.databricks_conn_id, **self.hook_params)
Review Comment:
By default, Connection will return `airflow.providers.databricks.hooks.databricks.DatabricksHook`.
And I didn't found the solution how to fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vincbeck commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
vincbeck commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1259567846
Where are you at @kazanzhy. Feel free to share the status of this PR, we might also be able to help you in case you need it so feel free to reach out :) It looks like your local modification reverted future.annotations made by Jarek. I encourage you to run the static-check locally with the command `breeze static-checks`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
eladkal commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1281315626
> I like the change - however completely removing the `execute` feels really aggressive. Shouldn't we deprecate notice and then remove later? This will break a _lot_ of workflows if anyone's dependency get's bumped without them knowing about the change.
This PR is backward compatible.
The operators are now inheriting from `SQLExecuteQueryOperator` so they use this operator `execute()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kazanzhy commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
kazanzhy commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1281378202
@fritz-astronomer I'll be so grateful if you will help me test these changes when RC will be ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1250394206
(and rebase the change after __future__.annotations change too :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #25717:
URL: https://github.com/apache/airflow/pull/25717#discussion_r985566382
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -162,6 +164,68 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
+ """
+ Executes SQL code in a specific database
+ :param sql: the SQL code or string pointing to a template file to be executed (templated).
+ File must have a '.sql' extensions.
+ :param autocommit: (optional) if True, each command is automatically committed (default: False).
+ :param parameters: (optional) the parameters to render the SQL query with.
+ :param handler: (optional) the function that will be applied to the cursor.
+ :param split_statements: (optional) if split single SQL string into statements (default: False).
+ :param return_last: (optional) if return the result of only last statement (default: True).
+ """
+
+ template_fields: Sequence[str] = ('sql', 'parameters')
+ template_ext: Sequence[str] = ('.sql',)
+ ui_color = '#cdaaed'
Review Comment:
Lets add
`template_fields_renderers = {"sql": "sql"}`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
syedahsn commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1303232314
Sure, I'll get started on that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] syedahsn commented on pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
syedahsn commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1303219879
This PR is causing the `RedshiftSQLOperator` to fail with the message
```
AttributeError: 'RedshiftSQLOperator' object has no attribute 'redshift_conn_id'
--
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/opt/airflow/airflow/models/taskinstance.py", line 1450, in _run_raw_task
self._execute_task_with_callbacks(context, test_mode)
File "/opt/airflow/airflow/models/taskinstance.py", line 1569, in _execute_task_with_callbacks
task_orig = self.render_templates(context=context)
File "/opt/airflow/airflow/models/taskinstance.py", line 2184, in render_templates
original_task.render_template_fields(context)
File "/opt/airflow/airflow/models/baseoperator.py", line 1196, in render_template_fields
self._do_render_template_fields(self, self.template_fields, context, jinja_env, set())
File "/opt/airflow/airflow/utils/session.py", line 75, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/airflow/airflow/models/abstractoperator.py", line 405, in _do_render_template_fields
f"{attr_name!r} is configured as a template field "
AttributeError: 'redshift_conn_id' is configured as a template field but RedshiftSQLOperator does not have this attribute.
```
The reason seems to be the fact that the `redshift_conn_id` attribute in the `RedshiftSQLOperator` was a tempate field, but now that it no longer exists, it is causing issues.
In terms of maintaining backward compatibility, should we introduce the attribute back to the `RedshiftSQLOperator` to resolve the issue? Or is there a better approach? I'd love to hear other opinions on this. Thanks!
@eladkal @kazanzhy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kazanzhy commented on pull request #25717: Add SQLExecuteQueryOperator (not tested)
Posted by GitBox <gi...@apache.org>.
kazanzhy commented on PR #25717:
URL: https://github.com/apache/airflow/pull/25717#issuecomment-1227836210
@vincbeck thanks for the suggestions.
Currently, I'm trying to fix tests, because before that each hook was patched by mock and now there're a lot of issues with that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] vincbeck commented on a diff in pull request #25717: Add SQLExecuteQueryOperator (not tested)
Posted by GitBox <gi...@apache.org>.
vincbeck commented on code in PR #25717:
URL: https://github.com/apache/airflow/pull/25717#discussion_r954257926
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -112,6 +124,61 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
Review Comment:
Dont forget to add documentation for this new operator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kazanzhy commented on a diff in pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
kazanzhy commented on code in PR #25717:
URL: https://github.com/apache/airflow/pull/25717#discussion_r985225576
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -162,6 +164,68 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
+ """
+ Executes SQL code in a specific database
+ :param sql: the SQL code or string pointing to a template file to be executed (templated).
+ File must have a '.sql' extensions.
+ :param autocommit: (optional) if True, each command is automatically committed (default: False).
+ :param parameters: (optional) the parameters to render the SQL query with.
+ :param handler: (optional) the function that will be applied to the cursor.
+ :param split_statements: (optional) if split single SQL string into statements (default: False).
+ :param return_last: (optional) if return the result of only last statement (default: True).
+ """
+
+ template_fields: Sequence[str] = ('sql', 'parameters')
+ template_ext: Sequence[str] = ('.sql',)
+ ui_color = '#cdaaed'
+
+ def __init__(
+ self,
+ *,
+ sql: str | list[str],
+ autocommit: bool = False,
+ parameters: Mapping | Iterable | None = None,
+ handler: Callable[[Any], Any] = fetch_all_handler,
+ split_statements: bool = False,
+ return_last: bool = True,
Review Comment:
Based on my previous PR which had broke compatibility I decided to avoid it here.
Talking about the `return_last` it works only when `do_xcom_push=True` (by default True) and `split_statements=True` (by default False).
Probably it should be renamed to `return_last_if_split_statements` :)
Why it's `True` by default?
I think when we added `split_statements` we wanted to keep the same returning type independent what were passed there.
https://github.com/apache/airflow/pull/23971#discussion_r922847638
If the user has in the legacy code, let's say, PostgresOperator. By default
it has `do_xcom_push = True`, `split_statements = False` and `return_last = True`.
https://github.com/apache/airflow/blob/69da98cdb194de3544368b6bd7c47dcc7ace8814/airflow/models/baseoperator.py#L735
https://github.com/apache/airflow/blob/69da98cdb194de3544368b6bd7c47dcc7ace8814/airflow/providers/common/sql/hooks/sql.py#L261
So if the's a string as an SQL parameter. then result of this query will be returned. If the list of strings is passed as an SQL parameter, then wil be returned list of results of each statement. Of course in case if handler were passed. I think here we discussed it here https://github.com/apache/airflow/pull/23971#discussion_r885115775
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a diff in pull request #25717: Add SQLExecuteQueryOperator
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #25717:
URL: https://github.com/apache/airflow/pull/25717#discussion_r985078769
##########
airflow/providers/common/sql/operators/sql.py:
##########
@@ -162,6 +164,68 @@ def get_db_hook(self) -> DbApiHook:
return self._hook
+class SQLExecuteQueryOperator(BaseSQLOperator):
+ """
+ Executes SQL code in a specific database
+ :param sql: the SQL code or string pointing to a template file to be executed (templated).
+ File must have a '.sql' extensions.
+ :param autocommit: (optional) if True, each command is automatically committed (default: False).
+ :param parameters: (optional) the parameters to render the SQL query with.
+ :param handler: (optional) the function that will be applied to the cursor.
+ :param split_statements: (optional) if split single SQL string into statements (default: False).
+ :param return_last: (optional) if return the result of only last statement (default: True).
+ """
+
+ template_fields: Sequence[str] = ('sql', 'parameters')
+ template_ext: Sequence[str] = ('.sql',)
+ ui_color = '#cdaaed'
+
+ def __init__(
+ self,
+ *,
+ sql: str | list[str],
+ autocommit: bool = False,
+ parameters: Mapping | Iterable | None = None,
+ handler: Callable[[Any], Any] = fetch_all_handler,
+ split_statements: bool = False,
+ return_last: bool = True,
Review Comment:
This parameter works only with `do_xcom_push=True`.
I have two thoughts on this:
1. If `return_last=True` should we under the hood set `do_xcom_push=True` ?
2. Assuming we go with (1) then probably the default of `return_last` should be False?
My goal here is to prevent the undesired state of `do_xcom_push=False` with `return_last=True`.
If user set `return_last=True` we know for sure xcom needs to be pushed.
if user set `return_last=False` we don't know what is the user wish for xcom so we need to also look for `do_xcom_push` value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org