potiuk commented on code in PR #72287:
URL: https://github.com/apache/airflow/pull/72287#discussion_r4062418559
##########
providers/http/tests/unit/http/triggers/test_http.py:
##########
@@ -36,6 +36,7 @@
HttpSensorTrigger,
HttpTrigger,
)
+from airflow.sdk.execution_time.context import AssetStateStoreAccessors
Review Comment:
This import is unguarded at module level, but `AssetStateStoreAccessors`
first shipped in Airflow 3.3.0. `PROVIDERS_COMPATIBILITY_TESTS_MATRIX` runs
this provider's unit tests against 2.11.1, 3.0.6, 3.1.8 and 3.2.2 as well, so
on those four the import raises at collection and every test in this file fails
— not just the new ones.
The Iceberg trigger's tests guard it by importing inside the version-gated
test instead
(`providers/apache/iceberg/tests/unit/apache/iceberg/triggers/test_iceberg.py:232`),
with `AIRFLOW_V_3_3_PLUS` from `tests_common.test_utils.version_compat`. Most
of the shapes below would also work with a plain `MagicMock()` store, which
needs no import at all.
##########
providers/http/docs/triggers.rst:
##########
@@ -66,27 +64,17 @@ Here's an example of using the ``HttpEventTrigger`` in an
``AssetWatcher`` to mo
}
- async def check_github_api_response(response):
+ async def check_github_api_response(response, asset_state_store=None):
"""Determine if a new version of Airflow has been released."""
- # Convert Variable.get to be asynchronous to be used on the Triggerer
- get_variable_async = sync_to_async(Variable.get)
-
- # Retrieve the previous and current release IDs (one from Variables,
one from response)
- previous_release_id = await get_variable_async(key="release_id_var",
default=None)
data = response.json()
release_id = str(data["id"])
- if release_id == previous_release_id:
- return False
-
- # Parse and persist Airflow release data to be used in the downstream
Task
- release_name = data["name"]
- release_html_url = data["html_url"]
- set_variable_async = sync_to_async(Variable.set)
-
- await set_variable_async(key="release_id_var", value=str(release_id))
- await set_variable_async(key="release_name_var", value=release_name)
- await set_variable_async(key="release_html_url_var",
value=release_html_url)
+ if asset_state_store is not None:
+ previous_release_id = asset_state_store.get("release_id", None)
Review Comment:
`AssetStateStoreAccessor.get`/`.set` do a synchronous
`SUPERVISOR_COMMS.send(...)` round-trip
(`task-sdk/src/airflow/sdk/execution_time/context.py:759`), and this callable
runs on the triggerer's shared event loop — so a user copying this example
stalls every other trigger in the process on each poll. The accessor's own
docstring says `aget`/`aset` "await instead of blocking the event loop".
`aget`/`aset` are not in a released version yet (3.3.0/3.3.1 only have the
blocking API), so the example needs the same shape the Kinesis trigger uses:
```python
if hasattr(asset_state_store, "aget"):
previous_release_id = await asset_state_store.aget("release_id", None)
else:
previous_release_id = await asyncio.to_thread(asset_state_store.get,
"release_id", None)
```
Same for the `.set` on line 77.
##########
providers/http/docs/triggers.rst:
##########
@@ -109,10 +97,7 @@ Here's an example of using the ``HttpEventTrigger`` in an
``AssetWatcher`` to mo
def check_airflow_releases():
@task()
def print_airflow_release_info():
- # Retrieve and output values persisted from the
``response_check_path`` function
- 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("A new Airflow release was detected by the HTTP event
trigger.")
Review Comment:
The downstream task no longer demonstrates anything — it prints a fixed
string, so `check_airflow_releases` is now just scaffolding. The old version
showed how to get the release name and URL to a downstream task, which is the
half of the story readers actually need.
Since the state store is reachable from task context too, writing
`release_name` / `release_html_url` alongside `release_id` in the check and
reading them back here would keep the example end-to-end.
##########
providers/http/tests/unit/http/triggers/test_http.py:
##########
@@ -388,3 +389,88 @@ async def test_trigger_on_post_with_data(
assert kwargs["data"] == TEST_DATA
assert kwargs["json"] is None
assert kwargs["params"] is None
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
Review Comment:
`expected_result` is `True` in all six cases, so that half of the
parametrize carries no information, and the actual variation is an `if`/`elif`
chain dispatching on a string id — which is a switch statement wearing a
parametrize costume.
`AGENTS.md` asks to "use `@pytest.mark.parametrize` for multiple similar
inputs — consolidate tests that only differ in input/expected values".
Parametrizing over the callables themselves drops the chain:
```python
async def _single_positional(resp):
return resp == "ok"
async def _default_second_arg(resp, threshold=5):
return resp == "ok" and threshold == 5
@pytest.mark.parametrize("check", [_single_positional, _default_second_arg,
...])
async def test_run_response_check_callable_shapes(self, event_trigger,
check):
...
```
The two shapes that need `mock_store` can close over a module-level
sentinel, or take it from the fixture.
--
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]
