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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
