apache / apache/airflow

Pass `AssetStateStoreAccessors` into `response_check_path` Callable in `HttpEventTrigger`

Open Beginner friendly
#70,074 1 comment 0 reactions 0 assignees View on GitHub
area:providers kind:feature provider:http
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.