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/05/28 07:04:33 UTC
[GitHub] [airflow] eladkal opened a new pull request, #23979: Add `PrestoToSlackOperator`
eladkal opened a new pull request, #23979:
URL: https://github.com/apache/airflow/pull/23979
Adding the funcitonality to run a single query against presto and send the result as slack message.
Similar to `SnowflakeToSlackOperator`
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragement file, named `{pr_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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] ephraimbuddy commented on a diff in pull request #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on code in PR #23979:
URL: https://github.com/apache/airflow/pull/23979#discussion_r888908113
##########
airflow/providers/presto/transfers/presto_to_slack.py:
##########
@@ -0,0 +1,143 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# 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
+
+from pandas import DataFrame
+from tabulate import tabulate
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.presto.hooks.presto import PrestoHook
+from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class PrestoToSlackOperator(BaseOperator):
+ """
+ Executes a single SQL statement in Presto and sends the results to Slack. The results of the query are
+ rendered into the 'slack_message' parameter as a Pandas dataframe using a JINJA variable called '{{
+ results_df }}'. The 'results_df' variable name can be changed by specifying a different
+ 'results_df_name' parameter. The Tabulate library is added to the JINJA environment as a filter to
+ allow the dataframe to be rendered nicely. For example, set 'slack_message' to {{ results_df |
+ tabulate(tablefmt="pretty", headers="keys") }} to send the results to Slack as an ascii rendered table.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:PrestoToSlackOperator`
+
+ :param sql: The SQL statement to execute on Presto (templated)
+ :param slack_message: The templated Slack message to send with the data returned from Presto.
+ You can use the default JINJA variable {{ results_df }} to access the pandas dataframe containing the
+ SQL results
+ :param presto_conn_id: destination presto connection
+ :param slack_conn_id: The connection id for Slack
+ :param results_df_name: The name of the JINJA template's dataframe variable, default is 'results_df'
+ :param parameters: The parameters to pass to the SQL query
+ :param slack_token: The token to use to authenticate to Slack. If this is not provided, the
+ 'webhook_token' attribute needs to be specified in the 'Extra' JSON field against the slack_conn_id
+ :param slack_channel: The channel to send message. Override default from Slack connection.
+ """
+
+ template_fields: Sequence[str] = ('sql', 'slack_message')
+ template_ext: Sequence[str] = ('.sql', '.jinja', '.j2')
+ template_fields_renderers = {"sql": "sql", "slack_message": "jinja"}
+ times_rendered = 0
+
+ def __init__(
+ self,
+ *,
+ sql: str,
+ slack_message: str,
+ presto_conn_id: str = 'presto_default',
+ slack_conn_id: str = 'slack_default',
+ results_df_name: str = 'results_df',
+ parameters: Optional[Union[Iterable, Mapping]] = None,
+ slack_token: Optional[str] = None,
+ slack_channel: Optional[str] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+
+ self.presto_conn_id = presto_conn_id
+ self.sql = sql
+ self.parameters = parameters
+ self.slack_conn_id = slack_conn_id
+ self.slack_token = slack_token
+ self.slack_message = slack_message
+ self.results_df_name = results_df_name
+ self.slack_channel = slack_channel
+
+ def _get_query_results(self) -> DataFrame:
+ presto_hook = self._get_presto_hook()
+
+ self.log.info('Running SQL query: %s', self.sql)
+ df = presto_hook.get_pandas_df(self.sql, parameters=self.parameters)
+ return df
+
+ def _render_and_send_slack_message(self, context, df) -> None:
+ # Put the dataframe into the context and render the JINJA template fields
+ context[self.results_df_name] = df
+ self.render_template_fields(context)
+
+ slack_hook = self._get_slack_hook()
+ self.log.info('Sending slack message: %s', self.slack_message)
+ slack_hook.execute()
+
+ def _get_presto_hook(self) -> PrestoHook:
+ return PrestoHook(presto_conn_id=self.presto_conn_id)
+
+ def _get_slack_hook(self) -> SlackWebhookHook:
+ return SlackWebhookHook(
+ http_conn_id=self.slack_conn_id,
+ message=self.slack_message,
+ webhook_token=self.slack_token,
+ slack_channel=self.slack_channel,
+ )
+
+ def render_template_fields(self, context, jinja_env=None) -> None:
+ # If this is the first render of the template fields, exclude slack_message from rendering since
+ # the presto results haven't been retrieved yet.
+ if self.times_rendered == 0:
+ fields_to_render: Iterable[str] = filter(lambda x: x != 'slack_message', self.template_fields)
+ else:
+ fields_to_render = self.template_fields
+
+ if not jinja_env:
+ jinja_env = self.get_template_env()
+
+ # Add the tabulate library into the JINJA environment
+ jinja_env.filters['tabulate'] = tabulate
+
+ self._do_render_template_fields(self, fields_to_render, context, jinja_env, set())
+ self.times_rendered += 1
+
+ def execute(self, context: 'Context') -> None:
+ if not isinstance(self.sql, str):
+ raise AirflowException("Expected 'sql' parameter should be a string.")
+ if self.sql is None or self.sql.strip() == "":
+ raise AirflowException("Expected 'sql' parameter is missing.")
+ if self.slack_message is None or self.slack_message.strip() == "":
+ raise AirflowException("Expected 'slack_message' parameter is missing.")
Review Comment:
```suggestion
if not self.sql.strip():
raise AirflowException("Expected 'sql' parameter is missing.")
if not self.slack_message.strip():
raise AirflowException("Expected 'slack_message' parameter is missing.")
```
Since we use `typing`, I think we should assume the user supplied str for these parameters and check for empty string instead. WDYT?
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
eladkal merged PR #23979:
URL: https://github.com/apache/airflow/pull/23979
--
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] josh-fell commented on a diff in pull request #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
josh-fell commented on code in PR #23979:
URL: https://github.com/apache/airflow/pull/23979#discussion_r889237549
##########
airflow/providers/presto/transfers/presto_to_slack.py:
##########
@@ -0,0 +1,141 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# 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
+
+from pandas import DataFrame
+from tabulate import tabulate
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.presto.hooks.presto import PrestoHook
+from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class PrestoToSlackOperator(BaseOperator):
+ """
+ Executes a single SQL statement in Presto and sends the results to Slack. The results of the query are
+ rendered into the 'slack_message' parameter as a Pandas dataframe using a JINJA variable called '{{
+ results_df }}'. The 'results_df' variable name can be changed by specifying a different
+ 'results_df_name' parameter. The Tabulate library is added to the JINJA environment as a filter to
+ allow the dataframe to be rendered nicely. For example, set 'slack_message' to {{ results_df |
+ tabulate(tablefmt="pretty", headers="keys") }} to send the results to Slack as an ascii rendered table.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:PrestoToSlackOperator`
+
+ :param sql: The SQL statement to execute on Presto (templated)
+ :param slack_message: The templated Slack message to send with the data returned from Presto.
+ You can use the default JINJA variable {{ results_df }} to access the pandas dataframe containing the
+ SQL results
+ :param presto_conn_id: destination presto connection
+ :param slack_conn_id: The connection id for Slack
+ :param results_df_name: The name of the JINJA template's dataframe variable, default is 'results_df'
+ :param parameters: The parameters to pass to the SQL query
+ :param slack_token: The token to use to authenticate to Slack. If this is not provided, the
+ 'webhook_token' attribute needs to be specified in the 'Extra' JSON field against the slack_conn_id
+ :param slack_channel: The channel to send message. Override default from Slack connection.
+ """
+
+ template_fields: Sequence[str] = ('sql', 'slack_message')
Review Comment:
WDYT about adding `slack_channel` as well? I was thinking maybe users would configure the Slack channel as an Airflow variable which could be accessed via Jinja.
##########
docs/apache-airflow-providers-presto/operators/transfer/presto_to_slack.rst:
##########
@@ -0,0 +1,37 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/operator:PrestoToSlackOperator:
+
+PrestoToSlackOperator
+========================
+
+Use the :class:`~airflow.providers.presto.transfers.presto_to_slack.presto_to_slack` to post messages to predefined Slack
+channels.
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+
+This operator will execute a custom query on a selected Presto table and publish a Slack message that can be formatted
+and contain the resulting dataset (e.g. ASCII formatted dataframe).
+
+An example usage of the PrestoToSlackOperator is as follows:
+
+.. exampleinclude:: /../../tests/system/providers/presto/example_presto_to_slack.py
+ :language: python
Review Comment:
```suggestion
:language: python
:indent: 4
```
##########
docs/apache-airflow-providers-presto/operators/transfer/presto_to_slack.rst:
##########
@@ -0,0 +1,37 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/operator:PrestoToSlackOperator:
+
+PrestoToSlackOperator
+========================
+
+Use the :class:`~airflow.providers.presto.transfers.presto_to_slack.presto_to_slack` to post messages to predefined Slack
+channels.
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+
+This operator will execute a custom query on a selected Presto table and publish a Slack message that can be formatted
Review Comment:
```suggestion
This operator will execute a custom query in Presto and publish a Slack message that can be formatted
```
The query can involve more than 1 Presto table right?
--
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] josh-fell commented on a diff in pull request #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
josh-fell commented on code in PR #23979:
URL: https://github.com/apache/airflow/pull/23979#discussion_r889240757
##########
docs/apache-airflow-providers-presto/operators/transfer/presto_to_slack.rst:
##########
@@ -0,0 +1,37 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/operator:PrestoToSlackOperator:
+
+PrestoToSlackOperator
+========================
+
+Use the :class:`~airflow.providers.presto.transfers.presto_to_slack.presto_to_slack` to post messages to predefined Slack
+channels.
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+
+This operator will execute a custom query on a selected Presto table and publish a Slack message that can be formatted
+and contain the resulting dataset (e.g. ASCII formatted dataframe).
+
+An example usage of the PrestoToSlackOperator is as follows:
+
+.. exampleinclude:: /../../tests/system/providers/presto/example_presto_to_slack.py
+ :language: python
Review Comment:
```suggestion
:language: python
:dedent: 4
```
##########
docs/apache-airflow-providers-presto/operators/transfer/presto_to_slack.rst:
##########
@@ -0,0 +1,37 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/operator:PrestoToSlackOperator:
+
+PrestoToSlackOperator
+========================
+
+Use the :class:`~airflow.providers.presto.transfers.presto_to_slack.presto_to_slack` to post messages to predefined Slack
+channels.
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+
+This operator will execute a custom query on a selected Presto table and publish a Slack message that can be formatted
+and contain the resulting dataset (e.g. ASCII formatted dataframe).
+
+An example usage of the PrestoToSlackOperator is as follows:
+
+.. exampleinclude:: /../../tests/system/providers/presto/example_presto_to_slack.py
+ :language: python
Review Comment:
Sorry `dedent` not `indent`
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
eladkal commented on PR #23979:
URL: https://github.com/apache/airflow/pull/23979#issuecomment-1146260469
Case 3 + 4 of [AIP 21](https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-21%3A+Changes+in+import+paths):
> In case of transfer operators where two providers are involved, the transfer operators will be moved to "source" of the transfer. When there is only one provider as target but source is a database or another non-provider source, the operator is put to the target provider.
From that description I think the source is the right choice
On a larger scope discussion - The fact that we both need to look up where operator should be is a problem. If we are not sure I think users are even more unsure where to look. I think it's time that we will consider a simpler rules like "it's will **always** be in the target name". e.g XtoY will always be in Y provider (Y is guaranteed to be exist while X is not). I will raise this in the mailing list as an modification to AIP-21 (just as soon as mailing list will be a bit more quite as there many parallel discussions right now).
Waiting for your confirmation before merging.
--
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] github-actions[bot] commented on pull request #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #23979:
URL: https://github.com/apache/airflow/pull/23979#issuecomment-1146237574
The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23979:
URL: https://github.com/apache/airflow/pull/23979#issuecomment-1146269044
> From that description I think the source is the right choice
This decision has been reverted later - see the last comment here: https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-8+Split+Providers+into+Separate+Packages+for+Airflow+2.0
And we documented it in Readme (AIPs often get outdated over time - that's why we move things to README to make sure they reflect the currrent state).
> On a larger scope discussion - The fact that we both need to look up where operator should be is a problem. If we are not sure I think users are even more unsure where to look. I think it's time that we will consider a simpler rules like "it's will always be in the target name". e.g XtoY will always be in Y provider (Y is guaranteed to be exist while X is not). I will raise this in the mailing list as an modification to AIP-21 (just as soon as mailing list will be a bit more quite as there many parallel discussions right now).
I think it's a bad idea. The maintainability rule was specifically foreseen for the "future" change where we split providers and people from given "provider" stakeholder should be able to subscribe to single repo/folder to see things that interest them for maintenance. This basically kills the idea of separating providers, if we are going to move "transfer" operators to source as AWS people will have to be more focused on all providers that "interface" with providers. The "maintainability" rule makes it fare more resaonable to split the providers.
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
eladkal commented on PR #23979:
URL: https://github.com/apache/airflow/pull/23979#issuecomment-1146281820
> And we documented it in Readme (AIPs often get outdated over time - that's why we move things to README to make sure they reflect the currrent state).
We are all for one source of truth :)
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23979:
URL: https://github.com/apache/airflow/pull/23979#issuecomment-1146235832
I am a bit torn on that one. I think it's fine to keep it is as is (in "presto" provider), but I have some doubt I wanted to share.
First thought is that this operator should really be part of "Slack" provider as this is the "Target". this is mostly what we chose for most "transfer" operators and there "target" is usualy more interested in "maintaining" the operator.
However in this case believe the "maintainability" rule favours Presto over Slack. Slack can be seen as utility/notification rather than "transfer" side.
https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#naming-conventions-for-provider-packages
> It is often debatable where to put transfer operators but we agreed to the following criteria:
> We use "maintainability" of the operators as the main criteria - so the transfer operator should be kept at the provider which has highest "interest" in the transfer operator
> For Cloud Providers or Service providers that usually means that the transfer operators should land at the "target" side of the transfer
Do you also see it that way @eladkal ?
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #23979:
URL: https://github.com/apache/airflow/pull/23979#discussion_r889246453
##########
docs/apache-airflow-providers-presto/operators/transfer/presto_to_slack.rst:
##########
@@ -0,0 +1,37 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _howto/operator:PrestoToSlackOperator:
+
+PrestoToSlackOperator
+========================
+
+Use the :class:`~airflow.providers.presto.transfers.presto_to_slack.presto_to_slack` to post messages to predefined Slack
+channels.
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+
+This operator will execute a custom query on a selected Presto table and publish a Slack message that can be formatted
Review Comment:
yes, It's a single SELECT statement. It can be with multiple tables.
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #23979:
URL: https://github.com/apache/airflow/pull/23979#discussion_r888918469
##########
airflow/providers/presto/transfers/presto_to_slack.py:
##########
@@ -0,0 +1,143 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# 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
+
+from pandas import DataFrame
+from tabulate import tabulate
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.presto.hooks.presto import PrestoHook
+from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class PrestoToSlackOperator(BaseOperator):
+ """
+ Executes a single SQL statement in Presto and sends the results to Slack. The results of the query are
+ rendered into the 'slack_message' parameter as a Pandas dataframe using a JINJA variable called '{{
+ results_df }}'. The 'results_df' variable name can be changed by specifying a different
+ 'results_df_name' parameter. The Tabulate library is added to the JINJA environment as a filter to
+ allow the dataframe to be rendered nicely. For example, set 'slack_message' to {{ results_df |
+ tabulate(tablefmt="pretty", headers="keys") }} to send the results to Slack as an ascii rendered table.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:PrestoToSlackOperator`
+
+ :param sql: The SQL statement to execute on Presto (templated)
+ :param slack_message: The templated Slack message to send with the data returned from Presto.
+ You can use the default JINJA variable {{ results_df }} to access the pandas dataframe containing the
+ SQL results
+ :param presto_conn_id: destination presto connection
+ :param slack_conn_id: The connection id for Slack
+ :param results_df_name: The name of the JINJA template's dataframe variable, default is 'results_df'
+ :param parameters: The parameters to pass to the SQL query
+ :param slack_token: The token to use to authenticate to Slack. If this is not provided, the
+ 'webhook_token' attribute needs to be specified in the 'Extra' JSON field against the slack_conn_id
+ :param slack_channel: The channel to send message. Override default from Slack connection.
+ """
+
+ template_fields: Sequence[str] = ('sql', 'slack_message')
+ template_ext: Sequence[str] = ('.sql', '.jinja', '.j2')
+ template_fields_renderers = {"sql": "sql", "slack_message": "jinja"}
+ times_rendered = 0
+
+ def __init__(
+ self,
+ *,
+ sql: str,
+ slack_message: str,
+ presto_conn_id: str = 'presto_default',
+ slack_conn_id: str = 'slack_default',
+ results_df_name: str = 'results_df',
+ parameters: Optional[Union[Iterable, Mapping]] = None,
+ slack_token: Optional[str] = None,
+ slack_channel: Optional[str] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+
+ self.presto_conn_id = presto_conn_id
+ self.sql = sql
+ self.parameters = parameters
+ self.slack_conn_id = slack_conn_id
+ self.slack_token = slack_token
+ self.slack_message = slack_message
+ self.results_df_name = results_df_name
+ self.slack_channel = slack_channel
+
+ def _get_query_results(self) -> DataFrame:
+ presto_hook = self._get_presto_hook()
+
+ self.log.info('Running SQL query: %s', self.sql)
+ df = presto_hook.get_pandas_df(self.sql, parameters=self.parameters)
+ return df
+
+ def _render_and_send_slack_message(self, context, df) -> None:
+ # Put the dataframe into the context and render the JINJA template fields
+ context[self.results_df_name] = df
+ self.render_template_fields(context)
+
+ slack_hook = self._get_slack_hook()
+ self.log.info('Sending slack message: %s', self.slack_message)
+ slack_hook.execute()
+
+ def _get_presto_hook(self) -> PrestoHook:
+ return PrestoHook(presto_conn_id=self.presto_conn_id)
+
+ def _get_slack_hook(self) -> SlackWebhookHook:
+ return SlackWebhookHook(
+ http_conn_id=self.slack_conn_id,
+ message=self.slack_message,
+ webhook_token=self.slack_token,
+ slack_channel=self.slack_channel,
+ )
+
+ def render_template_fields(self, context, jinja_env=None) -> None:
+ # If this is the first render of the template fields, exclude slack_message from rendering since
+ # the presto results haven't been retrieved yet.
+ if self.times_rendered == 0:
+ fields_to_render: Iterable[str] = filter(lambda x: x != 'slack_message', self.template_fields)
+ else:
+ fields_to_render = self.template_fields
+
+ if not jinja_env:
+ jinja_env = self.get_template_env()
+
+ # Add the tabulate library into the JINJA environment
+ jinja_env.filters['tabulate'] = tabulate
+
+ self._do_render_template_fields(self, fields_to_render, context, jinja_env, set())
+ self.times_rendered += 1
+
+ def execute(self, context: 'Context') -> None:
+ if not isinstance(self.sql, str):
+ raise AirflowException("Expected 'sql' parameter should be a string.")
+ if self.sql is None or self.sql.strip() == "":
+ raise AirflowException("Expected 'sql' parameter is missing.")
+ if self.slack_message is None or self.slack_message.strip() == "":
+ raise AirflowException("Expected 'slack_message' parameter is missing.")
Review Comment:
https://github.com/apache/airflow/pull/23979/commits/8e2f2917c1d2f43ad77accceeb12ab1afa8aebba
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
eladkal commented on PR #23979:
URL: https://github.com/apache/airflow/pull/23979#issuecomment-1145985967
rebased and modified example dag to AIP-47
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
eladkal commented on PR #23979:
URL: https://github.com/apache/airflow/pull/23979#issuecomment-1145782410
Appriciate review on this one :)
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #23979:
URL: https://github.com/apache/airflow/pull/23979#discussion_r889243405
##########
airflow/providers/presto/transfers/presto_to_slack.py:
##########
@@ -0,0 +1,141 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# 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
+
+from pandas import DataFrame
+from tabulate import tabulate
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.presto.hooks.presto import PrestoHook
+from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class PrestoToSlackOperator(BaseOperator):
+ """
+ Executes a single SQL statement in Presto and sends the results to Slack. The results of the query are
+ rendered into the 'slack_message' parameter as a Pandas dataframe using a JINJA variable called '{{
+ results_df }}'. The 'results_df' variable name can be changed by specifying a different
+ 'results_df_name' parameter. The Tabulate library is added to the JINJA environment as a filter to
+ allow the dataframe to be rendered nicely. For example, set 'slack_message' to {{ results_df |
+ tabulate(tablefmt="pretty", headers="keys") }} to send the results to Slack as an ascii rendered table.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:PrestoToSlackOperator`
+
+ :param sql: The SQL statement to execute on Presto (templated)
+ :param slack_message: The templated Slack message to send with the data returned from Presto.
+ You can use the default JINJA variable {{ results_df }} to access the pandas dataframe containing the
+ SQL results
+ :param presto_conn_id: destination presto connection
+ :param slack_conn_id: The connection id for Slack
+ :param results_df_name: The name of the JINJA template's dataframe variable, default is 'results_df'
+ :param parameters: The parameters to pass to the SQL query
+ :param slack_token: The token to use to authenticate to Slack. If this is not provided, the
+ 'webhook_token' attribute needs to be specified in the 'Extra' JSON field against the slack_conn_id
+ :param slack_channel: The channel to send message. Override default from Slack connection.
+ """
+
+ template_fields: Sequence[str] = ('sql', 'slack_message')
Review Comment:
sure will add
--
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 #23979: Add `PrestoToSlackOperator`
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23979:
URL: https://github.com/apache/airflow/pull/23979#issuecomment-1146273104
> This decision has been reverted later - see the last comment here: https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-8+Split+Providers+into+Separate+Packages+for+Airflow+2.0
BTW. I added a note to https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-21%3A+Changes+in+import+paths to reflect that decision 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