jroachgolf84 opened a new issue, #70074:
URL: https://github.com/apache/airflow/issues/70074

   ### 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)
   


-- 
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]

Reply via email to