Pass `AssetStateStoreAccessors` into `response_check_path` Callable in `HttpEventTrigger`
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### 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:
```python
...
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.
```python
...
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: https://github.com/apache/airflow/blob/471df2dc2201f90b0874c731dd6662e47a7b396d/providers/http/src/airflow/providers/http/triggers/http.py#L362-L368
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?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Read providers/http/src/airflow/providers/http/triggers/http.py, especially HttpEventTrigger._run_response_check and the response_check_path import flow. Update the callable invocation so the asset state accessor is available to response checks, then verify that the documented use case can retrieve and persist state through that argument.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend
- Issue type
- Feature
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 76/100