You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "karenbraganz (via GitHub)" <gi...@apache.org> on 2025/05/30 21:05:57 UTC
[PR] Create HttpEventTrigger for AssetWatcher library [airflow]
karenbraganz opened a new pull request, #51253:
URL: https://github.com/apache/airflow/pull/51253
<!--
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/
-->
<!-- Please keep an empty line above the dashes. -->
This HttpEventTrigger can be used for event-driven DAG scheduling in Airflow 3. This trigger asynchronously sends a request to an API. If the API response passes the conditions set by the user in the `response_check` callable, a DAG run will be triggered.
---
**^ 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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2261326472
##########
providers/http/docs/triggers.rst:
##########
@@ -0,0 +1,144 @@
+
+ .. 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.
+
+HTTP Event Trigger
+==================
+
+.. _howto/trigger:HttpEventTrigger:
+
+The ``HttpEventTrigger`` is an event-based trigger that monitors whether responses
+from an API meet the conditions set by the user in the ``response_check`` callable.
+
+It is especially useful for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "RNHTTR (via GitHub)" <gi...@apache.org>.
RNHTTR commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2257763964
##########
providers/http/docs/triggers.rst:
##########
@@ -0,0 +1,144 @@
+
+ .. 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.
+
+HTTP Event Trigger
+==================
+
+.. _howto/trigger:HttpEventTrigger:
+
+The ``HttpEventTrigger`` is an event-based trigger that monitors whether responses
+from an API meet the conditions set by the user in the ``response_check`` callable.
+
+It is especially useful for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
+enabling event-driven DAGs based on API responses.
+
+How It Works
+------------
+
+1. Sends requests to an API.
+2. Uses the callable at ``response_check_path`` to evaluate the API response.
+3. If the callable returns ``True``, a ``TriggerEvent`` is emitted. This will trigger DAGs using this ``AssetWatcher`` for scheduling.
+
+.. note::
+ This trigger requires **Airflow >= 3.0** due to dependencies on ``AssetWatcher`` and event-driven scheduling infrastructure.
+
+Usage Example with AssetWatcher
+-------------------------------
+
+Here's an example of using the HttpEventTrigger in an AssetWatcher to monitor the GitHub API for new Airflow releases.
+
+.. code-block:: python
+
+ from __future__ import annotations
+
+ import datetime
+ import os
+
+ from asgiref.sync import sync_to_async
+
+ from airflow.providers.http.triggers.http import HttpEventTrigger
+ from airflow.sdk import Asset, AssetWatcher, Variable, dag, task
+
+ # This token must be generated through GitHub and added as an environment variable
+ token = os.getenv("GITHUB_TOKEN")
+
+ headers = {
+ "Accept": "application/vnd.github+json",
+ "Authorization": f"Bearer {token}",
+ "X-GitHub-Api-Version": "2022-11-28",
+ }
+
+
+ async def check_github_api_response(response):
+ data = response.json()
+ release_id = str(data["id"])
+ get_variable_sync = sync_to_async(Variable.get)
+ previous_release_id = await get_variable_sync(key="release_id_var", default=None)
+ if release_id == previous_release_id:
+ return False
+ release_name = data["name"]
+ release_html_url = data["html_url"]
+ set_variable_sync = sync_to_async(Variable.set)
+ await set_variable_sync(key="release_id_var", value=str(release_id))
+ await set_variable_sync(key="release_name_var", value=release_name)
+ await set_variable_sync(key="release_html_url_var", value=release_html_url)
+ return True
+
+
+ trigger = HttpEventTrigger(
+ endpoint="repos/apache/airflow/releases/latest",
+ method="GET",
+ http_conn_id="http_default",
+ headers=headers,
+ response_check_path="dags.check_airflow_releases.check_github_api_response",
Review Comment:
In this example, is this path pointing to this file? If so, it makes sense to leave a comment indicating that (or the relative location of the response check)
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "RNHTTR (via GitHub)" <gi...@apache.org>.
RNHTTR commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2261155695
##########
providers/http/docs/triggers.rst:
##########
@@ -0,0 +1,144 @@
+
+ .. 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.
+
+HTTP Event Trigger
+==================
+
+.. _howto/trigger:HttpEventTrigger:
+
+The ``HttpEventTrigger`` is an event-based trigger that monitors whether responses
+from an API meet the conditions set by the user in the ``response_check`` callable.
+
+It is especially useful for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
Review Comment:
```suggestion
It is designed for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
```
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on PR #51253:
URL: https://github.com/apache/airflow/pull/51253#issuecomment-3157128805
Also tested this locally with the usage example that I provided in the docs and confirmed that it works.
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2257123043
##########
providers/http/docs/triggers.rst:
##########
@@ -0,0 +1,144 @@
+
+ .. 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.
+
+HTTP Event Trigger
+==================
+
+.. _howto/trigger:HttpEventTrigger:
+
+The ``HttpEventTrigger`` is an event-based trigger that monitors whether responses
+from an API meet the conditions set by the user in the ``response_check`` callable.
+
+It is especially useful for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
+enabling event-driven DAGs based on API responses.
+
+How It Works
+------------
+
+1. Sends requests to an API.
+2. Uses the callable at ``response_check_path`` to evaluate the API response.
+3. If the callable returns ``True``, a ``TriggerEvent`` is emitted. This will trigger DAGs using this ``AssetWatcher`` for scheduling.
+
+.. note::
+ This trigger requires **Airflow >= 3.0** due to dependencies on ``AssetWatcher`` and event-driven scheduling infrastructure.
+
+Usage Example with AssetWatcher
+-------------------------------
+
+Here's an example of using the HttpEventTrigger in an AssetWatcher to monitor the GitHub API for new Airflow releases.
+
+.. code-block:: python
+
+ from __future__ import annotations
+
+ import datetime
+ import os
+
+ from asgiref.sync import sync_to_async
+
+ from airflow.providers.http.triggers.http import HttpEventTrigger
+ from airflow.sdk import Asset, AssetWatcher, Variable, dag, task
+
+ # This token must be generated through GitHub and added as an environment variable
+ token = os.getenv("GITHUB_TOKEN")
+
+ headers = {
+ "Accept": "application/vnd.github+json",
+ "Authorization": f"Bearer {token}",
+ "X-GitHub-Api-Version": "2022-11-28",
+ }
+
+
+ async def check_github_api_response(response):
+ data = response.json()
+ release_id = str(data["id"])
+ get_variable_sync = sync_to_async(Variable.get)
+ previous_release_id = await get_variable_sync(key="release_id_var", default=None)
+ if release_id == previous_release_id:
+ return False
+ release_name = data["name"]
+ release_html_url = data["html_url"]
+ set_variable_sync = sync_to_async(Variable.set)
+ await set_variable_sync(key="release_id_var", value=str(release_id))
+ await set_variable_sync(key="release_name_var", value=release_name)
+ await set_variable_sync(key="release_html_url_var", value=release_html_url)
+ return True
+
+
+ trigger = HttpEventTrigger(
+ endpoint="repos/apache/airflow/releases/latest",
+ method="GET",
+ http_conn_id="http_default",
+ headers=headers,
+ response_check_path="dags.check_airflow_releases.check_github_api_response",
+ )
+
+ asset = Asset(
+ "airflow_releases_asset", watchers=[AssetWatcher(name="airflow_releases_watcher", trigger=trigger)]
+ )
+
+
+ @dag(start_date=datetime.datetime(2024, 10, 1), schedule=asset, catchup=False)
+ def check_airflow_releases():
+ @task()
+ def print_airflow_release_info():
+ release_name = Variable.get("release_name_var")
+ release_html_url = Variable.get("release_html_url_var")
+ print(f"{release_name} has been released. Check it out at {release_html_url}")
+
+ print_airflow_release_info()
+
+
+ check_airflow_releases()
+
+Parameters
+----------
+
+``http_conn_id``
+ http connection id that has the base API url i.e https://www.google.com/ and optional authentication credentials.
+ Default headers can also be specified in the Extra field in json format.
+
+``auth_type``
+ The auth type for the service
+
+``method``
+ the API method to be called
+
+``endpoint``
+ Endpoint to be called, i.e. ``resource/v1/query?``
+
+``headers``
+ Additional headers to be passed through as a dict
+
+``data``
+ Payload to be uploaded or request parameters
+
+``extra_options``
+ Additional kwargs to pass when creating a request.
+
+``response_check_path``
+ Path to callable that evaluates whether the API response passes the conditions set by the user to trigger DAGs
+
+
+Important Notes
+---------------
+
+1. The ``response_check_path`` must contain the path to an asynchronous callable. Synchronous callables will raise an exception.
+2. If no ``response_check_path`` is provided, the trigger will emit an event for every API response without any checks. This could trigger a high number of concurrent DAG runs, which could lock the Airflow database.
Review Comment:
It seems it is something we do not want, we do not we make `response_check_path` mandatory?
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "RNHTTR (via GitHub)" <gi...@apache.org>.
RNHTTR commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2261161972
##########
providers/http/src/airflow/providers/http/triggers/http.py:
##########
@@ -18,19 +18,32 @@
import asyncio
import base64
+import importlib
+import inspect
import pickle
+import sys
from collections.abc import AsyncIterator
from importlib import import_module
from typing import TYPE_CHECKING, Any
import aiohttp
import requests
+from asgiref.sync import sync_to_async
from requests.cookies import RequestsCookieJar
from requests.structures import CaseInsensitiveDict
from airflow.exceptions import AirflowException
from airflow.providers.http.hooks.http import HttpAsyncHook
-from airflow.triggers.base import BaseTrigger, TriggerEvent
+from airflow.providers.http.version_compat import AIRFLOW_V_3_0_PLUS
+
+if AIRFLOW_V_3_0_PLUS:
+ from airflow.triggers.base import BaseEventTrigger, BaseTrigger, TriggerEvent
+else:
+ from airflow.triggers.base import ( # type: ignore
+ BaseTrigger,
+ BaseTrigger as BaseEventTrigger,
+ TriggerEvent,
+ )
Review Comment:
Should it be...
```python
from airflow.triggers.base import BaseTrigger, TriggerEvent
if AIRFLOW_V_3_0_PLUS:
from airflow.triggers.base import BaseEventTrigger
else:
from airflow.triggers.base import BaseTrigger as BaseEventTrigger
```
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2258213098
##########
providers/http/docs/triggers.rst:
##########
@@ -0,0 +1,144 @@
+
+ .. 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.
+
+HTTP Event Trigger
+==================
+
+.. _howto/trigger:HttpEventTrigger:
+
+The ``HttpEventTrigger`` is an event-based trigger that monitors whether responses
+from an API meet the conditions set by the user in the ``response_check`` callable.
+
+It is especially useful for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
+enabling event-driven DAGs based on API responses.
+
+How It Works
+------------
+
+1. Sends requests to an API.
+2. Uses the callable at ``response_check_path`` to evaluate the API response.
+3. If the callable returns ``True``, a ``TriggerEvent`` is emitted. This will trigger DAGs using this ``AssetWatcher`` for scheduling.
+
+.. note::
+ This trigger requires **Airflow >= 3.0** due to dependencies on ``AssetWatcher`` and event-driven scheduling infrastructure.
+
+Usage Example with AssetWatcher
+-------------------------------
+
+Here's an example of using the HttpEventTrigger in an AssetWatcher to monitor the GitHub API for new Airflow releases.
+
+.. code-block:: python
+
+ from __future__ import annotations
+
+ import datetime
+ import os
+
+ from asgiref.sync import sync_to_async
+
+ from airflow.providers.http.triggers.http import HttpEventTrigger
+ from airflow.sdk import Asset, AssetWatcher, Variable, dag, task
+
+ # This token must be generated through GitHub and added as an environment variable
+ token = os.getenv("GITHUB_TOKEN")
+
+ headers = {
+ "Accept": "application/vnd.github+json",
+ "Authorization": f"Bearer {token}",
+ "X-GitHub-Api-Version": "2022-11-28",
+ }
+
+
+ async def check_github_api_response(response):
+ data = response.json()
+ release_id = str(data["id"])
+ get_variable_sync = sync_to_async(Variable.get)
+ previous_release_id = await get_variable_sync(key="release_id_var", default=None)
+ if release_id == previous_release_id:
+ return False
+ release_name = data["name"]
+ release_html_url = data["html_url"]
+ set_variable_sync = sync_to_async(Variable.set)
+ await set_variable_sync(key="release_id_var", value=str(release_id))
+ await set_variable_sync(key="release_name_var", value=release_name)
+ await set_variable_sync(key="release_html_url_var", value=release_html_url)
+ return True
+
+
+ trigger = HttpEventTrigger(
+ endpoint="repos/apache/airflow/releases/latest",
+ method="GET",
+ http_conn_id="http_default",
+ headers=headers,
+ response_check_path="dags.check_airflow_releases.check_github_api_response",
Review Comment:
I have added a comment indicating this.
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2260553502
##########
providers/http/src/airflow/providers/http/triggers/http.py:
##########
@@ -18,19 +18,32 @@
import asyncio
import base64
+import importlib
+import inspect
import pickle
+import sys
from collections.abc import AsyncIterator
from importlib import import_module
from typing import TYPE_CHECKING, Any
import aiohttp
import requests
+from asgiref.sync import sync_to_async
from requests.cookies import RequestsCookieJar
from requests.structures import CaseInsensitiveDict
from airflow.exceptions import AirflowException
from airflow.providers.http.hooks.http import HttpAsyncHook
-from airflow.triggers.base import BaseTrigger, TriggerEvent
+from airflow.providers.http.version_compat import AIRFLOW_V_3_0_PLUS
+
+if AIRFLOW_V_3_0_PLUS:
+ from airflow.triggers.base import BaseEventTrigger, BaseTrigger, TriggerEvent
+else:
+ from airflow.triggers.base import ( # type: ignore
+ BaseTrigger,
+ BaseTrigger as BaseEventTrigger,
+ TriggerEvent,
+ )
Review Comment:
@eladkal To ensure version compatibility for the Airflow 2 tests, I need to import `BaseTrigger` as `BaseEventTrigger`.
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on PR #51253:
URL: https://github.com/apache/airflow/pull/51253#issuecomment-3157113454
@vincbeck just added some documentation and an example. Let me know if you have any feedback.
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2258211507
##########
providers/http/docs/triggers.rst:
##########
@@ -0,0 +1,144 @@
+
+ .. 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.
+
+HTTP Event Trigger
+==================
+
+.. _howto/trigger:HttpEventTrigger:
+
+The ``HttpEventTrigger`` is an event-based trigger that monitors whether responses
+from an API meet the conditions set by the user in the ``response_check`` callable.
+
+It is especially useful for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
+enabling event-driven DAGs based on API responses.
+
+How It Works
+------------
+
+1. Sends requests to an API.
+2. Uses the callable at ``response_check_path`` to evaluate the API response.
+3. If the callable returns ``True``, a ``TriggerEvent`` is emitted. This will trigger DAGs using this ``AssetWatcher`` for scheduling.
+
+.. note::
+ This trigger requires **Airflow >= 3.0** due to dependencies on ``AssetWatcher`` and event-driven scheduling infrastructure.
+
+Usage Example with AssetWatcher
+-------------------------------
+
+Here's an example of using the HttpEventTrigger in an AssetWatcher to monitor the GitHub API for new Airflow releases.
+
+.. code-block:: python
+
+ from __future__ import annotations
+
+ import datetime
+ import os
+
+ from asgiref.sync import sync_to_async
+
+ from airflow.providers.http.triggers.http import HttpEventTrigger
+ from airflow.sdk import Asset, AssetWatcher, Variable, dag, task
+
+ # This token must be generated through GitHub and added as an environment variable
+ token = os.getenv("GITHUB_TOKEN")
+
+ headers = {
+ "Accept": "application/vnd.github+json",
+ "Authorization": f"Bearer {token}",
+ "X-GitHub-Api-Version": "2022-11-28",
+ }
+
+
+ async def check_github_api_response(response):
+ data = response.json()
+ release_id = str(data["id"])
+ get_variable_sync = sync_to_async(Variable.get)
+ previous_release_id = await get_variable_sync(key="release_id_var", default=None)
+ if release_id == previous_release_id:
+ return False
+ release_name = data["name"]
+ release_html_url = data["html_url"]
+ set_variable_sync = sync_to_async(Variable.set)
+ await set_variable_sync(key="release_id_var", value=str(release_id))
+ await set_variable_sync(key="release_name_var", value=release_name)
+ await set_variable_sync(key="release_html_url_var", value=release_html_url)
+ return True
+
+
+ trigger = HttpEventTrigger(
+ endpoint="repos/apache/airflow/releases/latest",
+ method="GET",
+ http_conn_id="http_default",
+ headers=headers,
+ response_check_path="dags.check_airflow_releases.check_github_api_response",
+ )
+
+ asset = Asset(
+ "airflow_releases_asset", watchers=[AssetWatcher(name="airflow_releases_watcher", trigger=trigger)]
+ )
+
+
+ @dag(start_date=datetime.datetime(2024, 10, 1), schedule=asset, catchup=False)
+ def check_airflow_releases():
+ @task()
+ def print_airflow_release_info():
+ release_name = Variable.get("release_name_var")
+ release_html_url = Variable.get("release_html_url_var")
+ print(f"{release_name} has been released. Check it out at {release_html_url}")
+
+ print_airflow_release_info()
+
+
+ check_airflow_releases()
+
+Parameters
+----------
+
+``http_conn_id``
+ http connection id that has the base API url i.e https://www.google.com/ and optional authentication credentials.
+ Default headers can also be specified in the Extra field in json format.
+
+``auth_type``
+ The auth type for the service
+
+``method``
+ the API method to be called
+
+``endpoint``
+ Endpoint to be called, i.e. ``resource/v1/query?``
+
+``headers``
+ Additional headers to be passed through as a dict
+
+``data``
+ Payload to be uploaded or request parameters
+
+``extra_options``
+ Additional kwargs to pass when creating a request.
+
+``response_check_path``
+ Path to callable that evaluates whether the API response passes the conditions set by the user to trigger DAGs
+
+
+Important Notes
+---------------
+
+1. The ``response_check_path`` must contain the path to an asynchronous callable. Synchronous callables will raise an exception.
+2. If no ``response_check_path`` is provided, the trigger will emit an event for every API response without any checks. This could trigger a high number of concurrent DAG runs, which could lock the Airflow database.
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2258103841
##########
providers/http/docs/triggers.rst:
##########
@@ -0,0 +1,144 @@
+
+ .. 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.
+
+HTTP Event Trigger
+==================
+
+.. _howto/trigger:HttpEventTrigger:
+
+The ``HttpEventTrigger`` is an event-based trigger that monitors whether responses
+from an API meet the conditions set by the user in the ``response_check`` callable.
+
+It is especially useful for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
+enabling event-driven DAGs based on API responses.
+
+How It Works
+------------
+
+1. Sends requests to an API.
+2. Uses the callable at ``response_check_path`` to evaluate the API response.
+3. If the callable returns ``True``, a ``TriggerEvent`` is emitted. This will trigger DAGs using this ``AssetWatcher`` for scheduling.
+
+.. note::
+ This trigger requires **Airflow >= 3.0** due to dependencies on ``AssetWatcher`` and event-driven scheduling infrastructure.
+
+Usage Example with AssetWatcher
+-------------------------------
+
+Here's an example of using the HttpEventTrigger in an AssetWatcher to monitor the GitHub API for new Airflow releases.
+
+.. code-block:: python
+
+ from __future__ import annotations
+
+ import datetime
+ import os
+
+ from asgiref.sync import sync_to_async
+
+ from airflow.providers.http.triggers.http import HttpEventTrigger
+ from airflow.sdk import Asset, AssetWatcher, Variable, dag, task
+
+ # This token must be generated through GitHub and added as an environment variable
+ token = os.getenv("GITHUB_TOKEN")
+
+ headers = {
+ "Accept": "application/vnd.github+json",
+ "Authorization": f"Bearer {token}",
+ "X-GitHub-Api-Version": "2022-11-28",
+ }
+
+
+ async def check_github_api_response(response):
+ data = response.json()
+ release_id = str(data["id"])
+ get_variable_sync = sync_to_async(Variable.get)
+ previous_release_id = await get_variable_sync(key="release_id_var", default=None)
+ if release_id == previous_release_id:
+ return False
+ release_name = data["name"]
+ release_html_url = data["html_url"]
+ set_variable_sync = sync_to_async(Variable.set)
+ await set_variable_sync(key="release_id_var", value=str(release_id))
+ await set_variable_sync(key="release_name_var", value=release_name)
+ await set_variable_sync(key="release_html_url_var", value=release_html_url)
+ return True
+
+
+ trigger = HttpEventTrigger(
+ endpoint="repos/apache/airflow/releases/latest",
+ method="GET",
+ http_conn_id="http_default",
+ headers=headers,
+ response_check_path="dags.check_airflow_releases.check_github_api_response",
+ )
+
+ asset = Asset(
+ "airflow_releases_asset", watchers=[AssetWatcher(name="airflow_releases_watcher", trigger=trigger)]
+ )
+
+
+ @dag(start_date=datetime.datetime(2024, 10, 1), schedule=asset, catchup=False)
+ def check_airflow_releases():
+ @task()
+ def print_airflow_release_info():
+ release_name = Variable.get("release_name_var")
+ release_html_url = Variable.get("release_html_url_var")
+ print(f"{release_name} has been released. Check it out at {release_html_url}")
+
+ print_airflow_release_info()
+
+
+ check_airflow_releases()
+
+Parameters
+----------
+
+``http_conn_id``
+ http connection id that has the base API url i.e https://www.google.com/ and optional authentication credentials.
+ Default headers can also be specified in the Extra field in json format.
+
+``auth_type``
+ The auth type for the service
+
+``method``
+ the API method to be called
+
+``endpoint``
+ Endpoint to be called, i.e. ``resource/v1/query?``
+
+``headers``
+ Additional headers to be passed through as a dict
+
+``data``
+ Payload to be uploaded or request parameters
+
+``extra_options``
+ Additional kwargs to pass when creating a request.
+
+``response_check_path``
+ Path to callable that evaluates whether the API response passes the conditions set by the user to trigger DAGs
+
+
+Important Notes
+---------------
+
+1. The ``response_check_path`` must contain the path to an asynchronous callable. Synchronous callables will raise an exception.
+2. If no ``response_check_path`` is provided, the trigger will emit an event for every API response without any checks. This could trigger a high number of concurrent DAG runs, which could lock the Airflow database.
Review Comment:
I can make it mandatory.
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "RNHTTR (via GitHub)" <gi...@apache.org>.
RNHTTR merged PR #51253:
URL: https://github.com/apache/airflow/pull/51253
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2259077348
##########
providers/http/src/airflow/providers/http/triggers/http.py:
##########
@@ -18,19 +18,32 @@
import asyncio
import base64
+import importlib
+import inspect
import pickle
+import sys
from collections.abc import AsyncIterator
from importlib import import_module
from typing import TYPE_CHECKING, Any
import aiohttp
import requests
+from asgiref.sync import sync_to_async
from requests.cookies import RequestsCookieJar
from requests.structures import CaseInsensitiveDict
from airflow.exceptions import AirflowException
from airflow.providers.http.hooks.http import HttpAsyncHook
-from airflow.triggers.base import BaseTrigger, TriggerEvent
+from airflow.providers.http.version_compat import AIRFLOW_V_3_0_PLUS
+
+if AIRFLOW_V_3_0_PLUS:
+ from airflow.triggers.base import BaseEventTrigger, BaseTrigger, TriggerEvent
+else:
+ from airflow.triggers.base import ( # type: ignore
+ BaseTrigger,
+ BaseTrigger as BaseEventTrigger,
+ TriggerEvent,
+ )
Review Comment:
Why do we need if/else for the same import path?
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "RNHTTR (via GitHub)" <gi...@apache.org>.
RNHTTR commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2261155695
##########
providers/http/docs/triggers.rst:
##########
@@ -0,0 +1,144 @@
+
+ .. 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.
+
+HTTP Event Trigger
+==================
+
+.. _howto/trigger:HttpEventTrigger:
+
+The ``HttpEventTrigger`` is an event-based trigger that monitors whether responses
+from an API meet the conditions set by the user in the ``response_check`` callable.
+
+It is especially useful for **Airflow 3.0+** in combination with the ``AssetWatcher`` system,
Review Comment:
```suggestion
It is designed for **Airflow 3.0+** to be used in combination with the ``AssetWatcher`` system,
```
--
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] Create HttpEventTrigger for AssetWatcher library [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on code in PR #51253:
URL: https://github.com/apache/airflow/pull/51253#discussion_r2261325369
##########
providers/http/src/airflow/providers/http/triggers/http.py:
##########
@@ -18,19 +18,32 @@
import asyncio
import base64
+import importlib
+import inspect
import pickle
+import sys
from collections.abc import AsyncIterator
from importlib import import_module
from typing import TYPE_CHECKING, Any
import aiohttp
import requests
+from asgiref.sync import sync_to_async
from requests.cookies import RequestsCookieJar
from requests.structures import CaseInsensitiveDict
from airflow.exceptions import AirflowException
from airflow.providers.http.hooks.http import HttpAsyncHook
-from airflow.triggers.base import BaseTrigger, TriggerEvent
+from airflow.providers.http.version_compat import AIRFLOW_V_3_0_PLUS
+
+if AIRFLOW_V_3_0_PLUS:
+ from airflow.triggers.base import BaseEventTrigger, BaseTrigger, TriggerEvent
+else:
+ from airflow.triggers.base import ( # type: ignore
+ BaseTrigger,
+ BaseTrigger as BaseEventTrigger,
+ TriggerEvent,
+ )
Review Comment:
I made the changes suggested by @RNHTTR
--
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