Description
In this example (https://airflow.apache.org/docs/apache-airflow-providers-http/stable/triggers.html), the asgiref.sync.sync_to_async function is used in tandem with Variable.get/set to retrieve and persist data used in response_check_path. It looks a bit like this:
...
async def check_github_api_response(response):
data = response.json()
release_id = str(data["id"])
# Using async Variable.get
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"]
# Using async Variable.set
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
...With the work done in AIP-103, AssetStateStoreAccessors is available to a BaseEventTrigger using the asset_state_store attribute. This allows for Triggers to read and persist state for an Asset.
I'd recommend updating the _run_response_check method in HttpEventTrigger to pass self.asset_state_store to the response_check function, like this. This would allow for HttpEventTrigger users to use the supported tooling to retrieving and persisting data.
...
async def _run_response_check(self, response) -> bool:
"""Run the response_check callable provided by the user."""
response_check = await self._import_from_response_check_path()
if not inspect.iscoroutinefunction(response_check):
raise AirflowException("The response_check callable is not asynchronous.")
check = await response_check(response, self.asset_state_store) # Make the change here
return check
...Permalink:
|
async def _run_response_check(self, response) -> bool: |
|
"""Run the response_check callable provided by the user.""" |
|
response_check = await self._import_from_response_check_path() |
|
if not inspect.iscoroutinefunction(response_check): |
|
raise AirflowException("The response_check callable is not asynchronous.") |
|
check = await response_check(response) |
|
return check |
This way
Use case/motivation
Rather than trying to use something like sync_to_async to use Variable.get/set, exposing the AssetStateStoreAccessors to response_check_path by adding an asset_state_store argument would allow for users to retrieve/persist state properly.
Related issues
No currently related issues.
Are you willing to submit a PR?
Code of Conduct
Description
In this example (https://airflow.apache.org/docs/apache-airflow-providers-http/stable/triggers.html), the
asgiref.sync.sync_to_asyncfunction is used in tandem withVariable.get/setto retrieve and persist data used inresponse_check_path. It looks a bit like this:With the work done in AIP-103,
AssetStateStoreAccessorsis available to aBaseEventTriggerusing theasset_state_storeattribute. This allows for Triggers to read and persist state for an Asset.I'd recommend updating the
_run_response_checkmethod inHttpEventTriggerto passself.asset_state_storeto theresponse_checkfunction, like this. This would allow forHttpEventTriggerusers to use the supported tooling to retrieving and persisting data.Permalink:
airflow/providers/http/src/airflow/providers/http/triggers/http.py
Lines 362 to 368 in 471df2d
This way
Use case/motivation
Rather than trying to use something like
sync_to_asyncto useVariable.get/set, exposing theAssetStateStoreAccessorstoresponse_check_pathby adding anasset_state_storeargument would allow for users to retrieve/persist state properly.Related issues
No currently related issues.
Are you willing to submit a PR?
Code of Conduct