You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dondaum (via GitHub)" <gi...@apache.org> on 2025/10/20 22:54:52 UTC
[PR] feat: add async discord notifier [airflow]
dondaum opened a new pull request, #56911:
URL: https://github.com/apache/airflow/pull/56911
<!--
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.
-->
<!--
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 an 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 an asynchronous version of `DiscordNotifier`.
related: https://github.com/apache/airflow/issues/55237
I sucessfully tested the `DiscordNotifier` with the following Dag:
```Python
from datetime import timedelta
from airflow import DAG
from airflow.sdk.definitions.deadline import AsyncCallback, DeadlineAlert, DeadlineReference
from airflow.providers.discord.notifications.discord import DiscordNotifier
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import task
with DAG(
dag_id="deadline_discord_alert_example",
deadline=DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval=timedelta(seconds=20),
callback=AsyncCallback(
DiscordNotifier,
kwargs={
"text": "🚨 Dag {{ dag_run.dag_id }} missed deadline at {{ deadline.deadline_time }}. DagRun: {{ dag_run }}",
},
),
),
):
c = EmptyOperator(task_id="example_task")
@task()
def wait():
import time
time.sleep(60*5)
c >> wait()
```
<img width="1968" height="542" alt="1" src="https://github.com/user-attachments/assets/307673d4-03d2-4069-a771-6c480afca3d3" />
-----------------
<img width="1796" height="662" alt="2" src="https://github.com/user-attachments/assets/7517c8b3-b96a-42b4-b20b-b991ad045434" />
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+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 newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3453605769
> we could add it to our open items list and change all providers to use the common-provider once https://github.com/apache/airflow/pull/57143 is merged.
I'm not thrilled with how many todos we're leaving hanging, but I guess that's life. Resolve the comments that are resolved and I'll make a pass tomorrow.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2447283081
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,70 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+import aiohttp
from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+from airflow.providers.discord.version_compat import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.discord.version_compat import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise AirflowException(
Review Comment:
Thank you for your review.
I didn't change this part of the code, I just moved it to another class.
But I am happy to change it.
Done.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2476846859
##########
providers/discord/src/airflow/providers/discord/utils/__init__.py:
##########
@@ -0,0 +1,40 @@
+# 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.
+#
+# NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE
+# OVERWRITTEN WHEN PREPARING DOCUMENTATION FOR THE PACKAGES.
+#
+# IF YOU WANT TO MODIFY THIS FILE, YOU SHOULD MODIFY THE TEMPLATE
+# `PROVIDER__INIT__PY_TEMPLATE.py.jinja2` IN the `dev/breeze/src/airflow_breeze/templates` DIRECTORY
+#
+from __future__ import annotations
+
+from airflow.providers.common.compat.sdk import BaseHook, Connection
+
+
Review Comment:
Exactly. Yes, it makes sense to merge #57143 first. Then I can remove this code here and use it from `airflow.providers.common.compat`.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3493117204
> > we could add it to our open items list and change all providers to use the common-provider once #57143 is merged.
>
> I'm not thrilled with how many todos we're leaving hanging, but I guess that's life. Resolve the comments that are resolved and I'll make a pass tomorrow.
Let's reduce our list of open items. I have opened a PR for the Slack provider for use `airflow.providers.common.compat `https://github.com/apache/airflow/pull/57894
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2541699569
##########
providers/discord/tests/unit/discord/notifications/test_discord.py:
##########
@@ -55,3 +55,20 @@ def test_discord_notifier_notify(mock_execute):
assert notifier.hook.message == "This is a test message"
assert notifier.hook.avatar_url == "https://example.com/avatar.png"
assert notifier.hook.tts is False
+
+
+@pytest.mark.asyncio
+@patch(
+ "airflow.providers.discord.notifications.discord.DiscordWebhookAsyncHook.execute",
+ new_callable=AsyncMock,
+)
+async def test_async_notifier(mock_async_hook):
+ notifier = DiscordNotifier(
+ discord_conn_id="my_discord_conn_id",
+ text="This is a test message",
+ username="test_user",
+ avatar_url="https://example.com/avatar.png",
+ tts=False,
+ )
+ await notifier.async_notify({})
+ mock_async_hook.assert_called_once()
Review Comment:
I got it know. Thanks. Changed 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "ramitkataria (via GitHub)" <gi...@apache.org>.
ramitkataria commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2446327030
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str | None = None,
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await sync_to_async(self.get_connection)(self.http_conn_id)
Review Comment:
I think we should use `await self.aget_connection(...)` instead so we don't create a new thread. I noticed this in some other places as well so I created a PR for the ones already in main: https://github.com/apache/airflow/pull/56791
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3495089453
> Let's reduce our list of open items. I have opened a PR for the Slack provider for use airflow.providers.common.compat https://github.com/apache/airflow/pull/57894
To be clear, that wasn't a jab at you by any means. Just in case it came off the wrong way. I greatly appreciate your work on these and hope I didn't imply otherwise.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3547165781
Will take a look tomorrow.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2446978081
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str | None = None,
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await sync_to_async(self.get_connection)(self.http_conn_id)
Review Comment:
Thank you for your review.
Oh, I see we have this method from the `BaseHook` now. Ah, I looked at it before and thought it used the same `sync_to_async(...)`, that's why I didn't want to use it.
But I missed that native asynchronous code can now be used for the case it can return a standard Airflow connection via `SUPERVISOR_COMMS`, which takes place here: https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/execution_time/secrets/execution_api.py#L97
I need to invest some time to understand this better.
Anyway, I changed 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2446985521
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,70 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+import aiohttp
+from asgiref.sync import sync_to_async
from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.discord.version_compat import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection, webhook_endpoint: str | None) -> str:
+ """
Review Comment:
Thank you for your review. I fixed both.
`ruff` removed the `Connection` object from the `version_compat` module because I didn't add it in `__all__`. I didn't notice that.
Thanks for recognizing 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2498338020
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +177,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str = "",
Review Comment:
Changed.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2497436302
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,69 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
Review Comment:
Can you add the params to the docstring?
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,69 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
Review Comment:
Can you add the params to the docstring? (here and elsewhere)
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2541262511
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,70 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, *, tts: bool, message: str, username: str | None, avatar_url: str | None
+ ) -> str:
+ """
+ Build a valid Discord JSON payload.
+
+ :param tts: Is a text-to-speech message
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :return: Discord payload (str) to send
+ """
+ if len(message) > 200:
Review Comment:
```suggestion
if len(message) > 2000:
```
##########
providers/discord/pyproject.toml:
##########
@@ -59,7 +59,7 @@ requires-python = ">=3.10"
# After you modify the dependencies, and rebuild your Breeze CI image with ``breeze ci-image build``
dependencies = [
"apache-airflow>=2.10.0",
- "apache-airflow-providers-common-compat>=1.8.0",
+ "apache-airflow-providers-common-compat>=1.9.0",
Review Comment:
What is this bumping for?
##########
providers/discord/tests/unit/discord/notifications/test_discord.py:
##########
@@ -55,3 +55,20 @@ def test_discord_notifier_notify(mock_execute):
assert notifier.hook.message == "This is a test message"
assert notifier.hook.avatar_url == "https://example.com/avatar.png"
assert notifier.hook.tts is False
+
+
+@pytest.mark.asyncio
+@patch(
+ "airflow.providers.discord.notifications.discord.DiscordWebhookAsyncHook.execute",
+ new_callable=AsyncMock,
+)
+async def test_async_notifier(mock_async_hook):
+ notifier = DiscordNotifier(
+ discord_conn_id="my_discord_conn_id",
+ text="This is a test message",
+ username="test_user",
+ avatar_url="https://example.com/avatar.png",
+ tts=False,
+ )
+ await notifier.async_notify({})
+ mock_async_hook.assert_called_once()
Review Comment:
```suggestion
mock_async_hook.mock_calls == [
...
]
```
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2447068075
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str | None = None,
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await sync_to_async(self.get_connection)(self.http_conn_id)
Review Comment:
Oh, but it has to be backwards compatible. I need to fix it similar to thishttps://github.com/apache/airflow/blob/main/providers/slack/src/airflow/providers/slack/utils/__init__.py#L128
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2498271251
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,69 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
Review Comment:
Done.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W merged PR #56911:
URL: https://github.com/apache/airflow/pull/56911
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2447243935
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,70 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+import aiohttp
from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+from airflow.providers.discord.version_compat import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.discord.version_compat import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise AirflowException(
Review Comment:
We should avoid using `AirflowException` as stated here https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#dont-raise-airflowexception-directly
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3547150627
Is there anything I can do to move forward with this 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2541270662
##########
providers/discord/tests/unit/discord/notifications/test_discord.py:
##########
@@ -55,3 +55,20 @@ def test_discord_notifier_notify(mock_execute):
assert notifier.hook.message == "This is a test message"
assert notifier.hook.avatar_url == "https://example.com/avatar.png"
assert notifier.hook.tts is False
+
+
+@pytest.mark.asyncio
+@patch(
+ "airflow.providers.discord.notifications.discord.DiscordWebhookAsyncHook.execute",
+ new_callable=AsyncMock,
+)
+async def test_async_notifier(mock_async_hook):
+ notifier = DiscordNotifier(
+ discord_conn_id="my_discord_conn_id",
+ text="This is a test message",
+ username="test_user",
+ avatar_url="https://example.com/avatar.png",
+ tts=False,
+ )
+ await notifier.async_notify({})
+ mock_async_hook.assert_called_once()
Review Comment:
```suggestion
mock_async_hook.mock_calls == [
calls(
# arguments passed to this mocked call
)
]
```
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2541299734
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,70 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, *, tts: bool, message: str, username: str | None, avatar_url: str | None
+ ) -> str:
+ """
+ Build a valid Discord JSON payload.
+
+ :param tts: Is a text-to-speech message
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :return: Discord payload (str) to send
+ """
+ if len(message) > 200:
Review Comment:
Good catch. Changed.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "KoviAnusha (via GitHub)" <gi...@apache.org>.
KoviAnusha commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2446899106
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,70 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+import aiohttp
+from asgiref.sync import sync_to_async
from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.discord.version_compat import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection, webhook_endpoint: str | None) -> str:
+ """
Review Comment:
The Connection import should come from airflow.models.connection, not version_compat, since that module doesn’t define it. Also, you can make conn type optional (Connection | None) and fix type hints for http_conn_id as str | None to satisfy mypy.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3488794951
I'll mark it as draft. But please let us know when it's ready again. Looking forward to 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2498297527
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,74 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
Review Comment:
let's keep 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3439835070
Looks like you have a handful of comments already, so I'll let things settle a bit before I review, but feel free to ping me when you are ready. As always, thanks for the work!
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3496056738
> > Let's reduce our list of open items. I have opened a PR for the Slack provider for use airflow.providers.common.compat #57894
>
> To be clear, that wasn't a jab at you by any means. Just in case it came off the wrong way. I greatly appreciate your work on these and hope I didn't imply otherwise.
Don't worry, I got it right. I like keeping things clean and I consider myself to have an intrinsic motivation to clean up my own things.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2498284945
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,74 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
+ ) -> str:
+ """
+ Build a valid Discord JSON payload.
+
+ :param tts: Is a text-to-speech message
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :return: Discord payload (str) to send
+ """
+ payload: dict[str, Any] = {}
+
+ if username:
+ payload["username"] = username
+ if avatar_url:
+ payload["avatar_url"] = avatar_url
+
+ payload["tts"] = tts
+
+ if len(message) <= 2000:
+ payload["content"] = message
+ else:
+ raise ValueError("Discord message length must be 2000 or fewer characters.")
Review Comment:
Thank you. Changed 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2541589974
##########
providers/discord/tests/unit/discord/notifications/test_discord.py:
##########
@@ -55,3 +55,20 @@ def test_discord_notifier_notify(mock_execute):
assert notifier.hook.message == "This is a test message"
assert notifier.hook.avatar_url == "https://example.com/avatar.png"
assert notifier.hook.tts is False
+
+
+@pytest.mark.asyncio
+@patch(
+ "airflow.providers.discord.notifications.discord.DiscordWebhookAsyncHook.execute",
+ new_callable=AsyncMock,
+)
+async def test_async_notifier(mock_async_hook):
+ notifier = DiscordNotifier(
+ discord_conn_id="my_discord_conn_id",
+ text="This is a test message",
+ username="test_user",
+ avatar_url="https://example.com/avatar.png",
+ tts=False,
+ )
+ await notifier.async_notify({})
+ mock_async_hook.assert_called_once()
Review Comment:
just updated the example suggestion
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2544826151
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ *,
+ http_conn_id: str = "",
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await get_async_connection(self.http_conn_id)
+ return self.handler.get_webhook_endpoint(conn, self.webhook_endpoint)
+
+ async def execute(self) -> None:
+ """Execute the Discord webhook call."""
+ webhook_endpoint = await self._get_webhook_endpoint()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
+
+ session_kwargs: dict[str, Any] = {}
+ if self.proxy:
+ session_kwargs["proxy"] = self.proxy
+
+ async with aiohttp.ClientSession(**session_kwargs) as session:
Review Comment:
Thanks for the comment. Yes you are right. I checked `aiohttp.ClientSession` and `proxy` has `proxy: Optional[StrOrURL] = None`. I changed 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2498284151
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,74 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
Review Comment:
I changed it to “Keyword Only.” That's a good approach. Thanks.
I can rename `message` to `content`, but I think we then also need to rename the `hooks` init parameters, and most likely those for the operator and notfier. Yes, I could make it backwards compatible but I'm not sure it's worth the effort.
It's okay for me to keep it or change it.
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3487263126
Merged https://github.com/apache/airflow/pull/57143, let's get this one wrapped 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2498284151
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,74 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
Review Comment:
I changed it to “keyword only.” That's a good approach. Thanks.
I can rename `message` to `content`, but I think we then also need to rename the `hooks` init parameters, and most likely those for the operator and notfier. Yes, I could make it backwards compatible but I'm not sure it's worth the effort.
It's okay for me to keep it or change it.
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2447068075
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str | None = None,
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await sync_to_async(self.get_connection)(self.http_conn_id)
Review Comment:
Oh, but it has to be backwards compatible. I need to fix it similar to this https://github.com/apache/airflow/blob/main/providers/slack/src/airflow/providers/slack/utils/__init__.py#L128
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2447260024
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str | None = None,
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await sync_to_async(self.get_connection)(self.http_conn_id)
Review Comment:
IMO there are two options. Either we need to pin all providers that use `await self.aget_connection(...)` to the Airflow Task SDK version that has this backwards compatible change, or we need to implement similar functionality in the relevant providers. The last option makes more sense because using the Airflow task SDK forces Airflow 3.
WDYT ?
This is also very relevant for https://github.com/apache/airflow/pull/56791
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2476038118
##########
providers/discord/src/airflow/providers/discord/utils/__init__.py:
##########
@@ -0,0 +1,40 @@
+# 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.
+#
+# NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE
+# OVERWRITTEN WHEN PREPARING DOCUMENTATION FOR THE PACKAGES.
+#
+# IF YOU WANT TO MODIFY THIS FILE, YOU SHOULD MODIFY THE TEMPLATE
+# `PROVIDER__INIT__PY_TEMPLATE.py.jinja2` IN the `dev/breeze/src/airflow_breeze/templates` DIRECTORY
+#
+from __future__ import annotations
+
+from airflow.providers.common.compat.sdk import BaseHook, Connection
+
+
Review Comment:
This will be superseded by your other PR (https://github.com/apache/airflow/pull/57143/files), right? Should we merge that one first and update this one before merging 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2454411310
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str | None = None,
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await sync_to_async(self.get_connection)(self.http_conn_id)
Review Comment:
[PR](https://github.com/apache/airflow/pull/57143)
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "ramitkataria (via GitHub)" <gi...@apache.org>.
ramitkataria commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2453608665
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str | None = None,
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await sync_to_async(self.get_connection)(self.http_conn_id)
Review Comment:
Yea, makes sense. I also just noticed this: https://github.com/apache/airflow/issues/57018
So I was thinking we add the if statement with version check in the common provider and then all the other providers can just make a simple call
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2495962743
##########
providers/discord/src/airflow/providers/discord/utils/__init__.py:
##########
@@ -0,0 +1,40 @@
+# 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.
+#
+# NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE
+# OVERWRITTEN WHEN PREPARING DOCUMENTATION FOR THE PACKAGES.
+#
+# IF YOU WANT TO MODIFY THIS FILE, YOU SHOULD MODIFY THE TEMPLATE
+# `PROVIDER__INIT__PY_TEMPLATE.py.jinja2` IN the `dev/breeze/src/airflow_breeze/templates` DIRECTORY
+#
+from __future__ import annotations
+
+from airflow.providers.common.compat.sdk import BaseHook, Connection
+
+
Review Comment:
Done.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3493123722
Thank you for all the patience and support. I'm opening it up for another round of reviews.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2541306843
##########
providers/discord/tests/unit/discord/notifications/test_discord.py:
##########
@@ -55,3 +55,20 @@ def test_discord_notifier_notify(mock_execute):
assert notifier.hook.message == "This is a test message"
assert notifier.hook.avatar_url == "https://example.com/avatar.png"
assert notifier.hook.tts is False
+
+
+@pytest.mark.asyncio
+@patch(
+ "airflow.providers.discord.notifications.discord.DiscordWebhookAsyncHook.execute",
+ new_callable=AsyncMock,
+)
+async def test_async_notifier(mock_async_hook):
+ notifier = DiscordNotifier(
+ discord_conn_id="my_discord_conn_id",
+ text="This is a test message",
+ username="test_user",
+ avatar_url="https://example.com/avatar.png",
+ tts=False,
+ )
+ await notifier.async_notify({})
+ mock_async_hook.assert_called_once()
Review Comment:
I think I don't understand your suggested change here. Can you please explain?
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3453554038
CI is green now.
In my opinion, it makes sense to continue with this feature if it looks good.
Yes, the backwards compatible part is redundant with https://github.com/apache/airflow/pull/57143, but in my opinion we could add it to our open items list and change all providers to use the `common-provider` once https://github.com/apache/airflow/pull/57143 is merged.
WDYT ? @ferruzzi @ramitkataria
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2498199826
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,74 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
+ ) -> str:
+ """
+ Build a valid Discord JSON payload.
+
+ :param tts: Is a text-to-speech message
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :return: Discord payload (str) to send
+ """
+ payload: dict[str, Any] = {}
+
+ if username:
+ payload["username"] = username
+ if avatar_url:
+ payload["avatar_url"] = avatar_url
+
+ payload["tts"] = tts
+
+ if len(message) <= 2000:
+ payload["content"] = message
+ else:
+ raise ValueError("Discord message length must be 2000 or fewer characters.")
Review Comment:
```suggestion
if len(message) > 200:
raise ValueError("Discord message length must be 2000 or fewer characters.")
payload: dict[str, Any] = {
"content": message,
"tts": tts,
}
if username:
payload["username"] = username
if avatar_url:
payload["avatar_url"] = avatar_url
```
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,74 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
Review Comment:
let's make it keyword only, the order of these parameters does not matter or mean something
```suggestion
self, *, tts: bool, message: str, username: str | None, avatar_url: str | None
```
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -19,10 +19,74 @@
import json
import re
-from typing import Any
+from typing import TYPE_CHECKING, Any
-from airflow.exceptions import AirflowException
-from airflow.providers.http.hooks.http import HttpHook
+import aiohttp
+
+from airflow.providers.common.compat.connection import get_async_connection
+from airflow.providers.http.hooks.http import HttpAsyncHook, HttpHook
+
+if TYPE_CHECKING:
+ from airflow.providers.common.compat.sdk import Connection
+
+
+class DiscordCommonHandler:
+ """Contains the common functionality."""
+
+ def get_webhook_endpoint(self, conn: Connection | None, webhook_endpoint: str | None) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param conn: Airflow Discord connection
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ if webhook_endpoint:
+ endpoint = webhook_endpoint
+ elif conn:
+ extra = conn.extra_dejson
+ endpoint = extra.get("webhook_endpoint", "")
+ else:
+ raise ValueError(
+ "Cannot get webhook endpoint: No valid Discord webhook endpoint or http_conn_id supplied."
+ )
+
+ # make sure endpoint matches the expected Discord webhook format
+ if not re.fullmatch("webhooks/[0-9]+/[a-zA-Z0-9_-]+", endpoint):
+ raise ValueError(
+ 'Expected Discord webhook endpoint in the form of "webhooks/{webhook.id}/{webhook.token}".'
+ )
+
+ return endpoint
+
+ def build_discord_payload(
+ self, tts: bool, message: str, username: str | None, avatar_url: str | None
Review Comment:
do we want to rename `message` as `content` for consistency?
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +177,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str = "",
Review Comment:
```suggestion
*,
http_conn_id: str = "",
```
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2541324464
##########
providers/discord/pyproject.toml:
##########
@@ -59,7 +59,7 @@ requires-python = ">=3.10"
# After you modify the dependencies, and rebuild your Breeze CI image with ``breeze ci-image build``
dependencies = [
"apache-airflow>=2.10.0",
- "apache-airflow-providers-common-compat>=1.8.0",
+ "apache-airflow-providers-common-compat>=1.9.0",
Review Comment:
This change depends on `apache-airflow-providers-common-compat>=1.9.0`. It was published a few days ago. See
https://github.com/apache/airflow/pull/57143
https://github.com/apache/airflow/issues/58315
But for this reason [this](https://github.com/apache/airflow/pull/58393) I have to add `"apache-airflow-providers-common-compat>=1.8.0", # use next version`.
Changed.
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3552452090
I'll keep it open for a few days so that other can take a look. But overall looks good! Thanks!
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2542675335
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ *,
+ http_conn_id: str = "",
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await get_async_connection(self.http_conn_id)
+ return self.handler.get_webhook_endpoint(conn, self.webhook_endpoint)
+
+ async def execute(self) -> None:
+ """Execute the Discord webhook call."""
+ webhook_endpoint = await self._get_webhook_endpoint()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
+
+ session_kwargs: dict[str, Any] = {}
+ if self.proxy:
+ session_kwargs["proxy"] = self.proxy
+
+ async with aiohttp.ClientSession(**session_kwargs) as session:
Review Comment:
It seems we can set the `proxy` kwarg (and `proxies` for sync one) directly.
```suggestion
async with aiohttp.ClientSession(proxy=self.proxy) as session:
```
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2544019466
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ *,
+ http_conn_id: str = "",
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await get_async_connection(self.http_conn_id)
+ return self.handler.get_webhook_endpoint(conn, self.webhook_endpoint)
+
+ async def execute(self) -> None:
+ """Execute the Discord webhook call."""
+ webhook_endpoint = await self._get_webhook_endpoint()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
+
+ session_kwargs: dict[str, Any] = {}
+ if self.proxy:
+ session_kwargs["proxy"] = self.proxy
+
+ async with aiohttp.ClientSession(**session_kwargs) as session:
Review Comment:
looks like right. After @dondaum confirmed, I think we can merge 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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #56911:
URL: https://github.com/apache/airflow/pull/56911#issuecomment-3552174516
Approve as the remaining one is a nitpick. @dondaum, please click resolve for the comments you think you've already resolved :)
--
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
Re: [PR] feat: add async discord notifier [airflow]
Posted by "dondaum (via GitHub)" <gi...@apache.org>.
dondaum commented on code in PR #56911:
URL: https://github.com/apache/airflow/pull/56911#discussion_r2454305405
##########
providers/discord/src/airflow/providers/discord/hooks/discord_webhook.py:
##########
@@ -148,11 +173,100 @@ def execute(self) -> None:
# we only need https proxy for Discord
proxies = {"https": self.proxy}
- discord_payload = self._build_discord_payload()
+ discord_payload = self.handler.build_discord_payload(
+ tts=self.tts, message=self.message, username=self.username, avatar_url=self.avatar_url
+ )
self.run(
endpoint=self.webhook_endpoint,
data=discord_payload,
headers={"Content-type": "application/json"},
extra_options={"proxies": proxies},
)
+
+
+class DiscordWebhookAsyncHook(HttpAsyncHook):
+ """
+ This hook allows you to post messages to Discord using incoming webhooks using async HTTP.
+
+ Takes a Discord connection ID with a default relative webhook endpoint. The
+ default endpoint can be overridden using the webhook_endpoint parameter
+ (https://discordapp.com/developers/docs/resources/webhook).
+
+ Each Discord webhook can be pre-configured to use a specific username and
+ avatar_url. You can override these defaults in this hook.
+
+ :param http_conn_id: Http connection ID with host as "https://discord.com/api/" and
+ default webhook endpoint in the extra field in the form of
+ {"webhook_endpoint": "webhooks/{webhook.id}/{webhook.token}"}
+ :param webhook_endpoint: Discord webhook endpoint in the form of
+ "webhooks/{webhook.id}/{webhook.token}"
+ :param message: The message you want to send to your Discord channel
+ (max 2000 characters)
+ :param username: Override the default username of the webhook
+ :param avatar_url: Override the default avatar of the webhook
+ :param tts: Is a text-to-speech message
+ :param proxy: Proxy to use to make the Discord webhook call
+ """
+
+ default_headers = {
+ "Content-Type": "application/json",
+ }
+ conn_name_attr = "http_conn_id"
+ default_conn_name = "discord_default"
+ conn_type = "discord"
+ hook_name = "Async Discord"
+
+ def __init__(
+ self,
+ http_conn_id: str | None = None,
+ webhook_endpoint: str | None = None,
+ message: str = "",
+ username: str | None = None,
+ avatar_url: str | None = None,
+ tts: bool = False,
+ proxy: str | None = None,
+ *args: Any,
+ **kwargs: Any,
+ ) -> None:
+ super().__init__(*args, **kwargs)
+ self.http_conn_id = http_conn_id
+ self.webhook_endpoint = webhook_endpoint
+ self.message = message
+ self.username = username
+ self.avatar_url = avatar_url
+ self.tts = tts
+ self.proxy = proxy
+ self.handler = DiscordCommonHandler()
+
+ async def _get_webhook_endpoint(self) -> str:
+ """
+ Return the default webhook endpoint or override if a webhook_endpoint is manually supplied.
+
+ :param http_conn_id: The provided connection ID
+ :param webhook_endpoint: The manually provided webhook endpoint
+ :return: Webhook endpoint (str) to use
+ """
+ conn = None
+ if not self.webhook_endpoint and self.http_conn_id:
+ conn = await sync_to_async(self.get_connection)(self.http_conn_id)
Review Comment:
Ah, exactly, there is now also a common provider.
Yes, we should implement it there. That's a good idea.
Let me extract this functionality [here ](https://github.com/dondaum/airflow/blob/7e08be77048ef53acf2d35940e9d02d03740aaa5/providers/discord/src/airflow/providers/discord/version_compat.py#L54) and move it to the shared provider.
--
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