pierrejeambrun commented on code in PR #54652:
URL: https://github.com/apache/airflow/pull/54652#discussion_r2315766793
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py:
##########
@@ -81,5 +87,11 @@ def next_run_assets(
)
]
+ for ev in events:
+ queued = ev.get("queued")
+ if not queued:
+ ev["lastUpdate"] = None
+ ev.pop("queued", None)
+
Review Comment:
I think this could be incorporated directly above in the query when we
construct the `lastUpdate` column.
This will avoid having to iterate over the events again and poping `queued`.
We just need to combine those two:
```
func.max(AssetEvent.timestamp).label("lastUpdate"),
func.max(
case(
(AssetDagRunQueue.asset_id.is_not(None), 1),
else_=0,
)
).label("queued"),
```
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py:
##########
@@ -74,3 +77,128 @@ def test_should_respond_401(self,
unauthenticated_test_client):
def test_should_respond_403(self, unauthorized_test_client):
response = unauthorized_test_client.get("/next_run_assets/upstream")
assert response.status_code == 403
+
+ def test_should_set_last_update_only_for_queued_and_hide_flag(self,
test_client, dag_maker):
+ with dag_maker(
+ dag_id="two_assets_equal",
+ schedule=[
+ Asset(uri="s3://bucket/A", name="A"),
+ Asset(uri="s3://bucket/B", name="B"),
+ ],
+ serialized=True,
+ ):
+ EmptyOperator(task_id="t")
+
+ dr = dag_maker.create_dagrun()
+ dag_maker.sync_dagbag_to_db()
+
+ with create_session() as session:
+ assets = {
+ a.uri: a
+ for a in session.query(AssetModel).filter(
+ AssetModel.uri.in_(["s3://bucket/A", "s3://bucket/B"])
+ )
+ }
+ # Queue and add an event only for A
+ session.add(
+ AssetDagRunQueue(asset_id=assets["s3://bucket/A"].id,
target_dag_id="two_assets_equal")
+ )
+ session.add(
+ AssetEvent(asset_id=assets["s3://bucket/A"].id,
timestamp=dr.logical_date or pendulum.now())
+ )
+ session.commit()
+
+ response = test_client.get("/next_run_assets/two_assets_equal")
+ assert response.status_code == 200
+ assert response.json() == {
+ "asset_expression": {
+ "all": [
+ {
+ "asset": {
+ "uri": "s3://bucket/A",
+ "name": "A",
+ "group": "asset",
+ "id": mock.ANY,
+ }
+ },
+ {
+ "asset": {
+ "uri": "s3://bucket/B",
+ "name": "B",
+ "group": "asset",
+ "id": mock.ANY,
+ }
+ },
+ ]
+ },
+ # events are ordered by uri
+ "events": [
+ {"id": mock.ANY, "uri": "s3://bucket/A", "name": "A",
"lastUpdate": mock.ANY},
+ {"id": mock.ANY, "uri": "s3://bucket/B", "name": "B",
"lastUpdate": None},
+ ],
+ }
+
+ def test_two_assets_only_one_queued_last_update_norm(self, test_client,
dag_maker):
Review Comment:
you can use the `session` fixture instead.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py:
##########
@@ -74,3 +77,128 @@ def test_should_respond_401(self,
unauthenticated_test_client):
def test_should_respond_403(self, unauthorized_test_client):
response = unauthorized_test_client.get("/next_run_assets/upstream")
assert response.status_code == 403
+
+ def test_should_set_last_update_only_for_queued_and_hide_flag(self,
test_client, dag_maker):
+ with dag_maker(
+ dag_id="two_assets_equal",
+ schedule=[
+ Asset(uri="s3://bucket/A", name="A"),
+ Asset(uri="s3://bucket/B", name="B"),
+ ],
+ serialized=True,
+ ):
+ EmptyOperator(task_id="t")
+
+ dr = dag_maker.create_dagrun()
+ dag_maker.sync_dagbag_to_db()
+
+ with create_session() as session:
+ assets = {
+ a.uri: a
+ for a in session.query(AssetModel).filter(
+ AssetModel.uri.in_(["s3://bucket/A", "s3://bucket/B"])
+ )
+ }
+ # Queue and add an event only for A
+ session.add(
+ AssetDagRunQueue(asset_id=assets["s3://bucket/A"].id,
target_dag_id="two_assets_equal")
+ )
+ session.add(
+ AssetEvent(asset_id=assets["s3://bucket/A"].id,
timestamp=dr.logical_date or pendulum.now())
+ )
+ session.commit()
+
+ response = test_client.get("/next_run_assets/two_assets_equal")
+ assert response.status_code == 200
+ assert response.json() == {
+ "asset_expression": {
+ "all": [
+ {
+ "asset": {
+ "uri": "s3://bucket/A",
+ "name": "A",
+ "group": "asset",
+ "id": mock.ANY,
+ }
+ },
+ {
+ "asset": {
+ "uri": "s3://bucket/B",
+ "name": "B",
+ "group": "asset",
+ "id": mock.ANY,
+ }
+ },
+ ]
+ },
+ # events are ordered by uri
+ "events": [
+ {"id": mock.ANY, "uri": "s3://bucket/A", "name": "A",
"lastUpdate": mock.ANY},
+ {"id": mock.ANY, "uri": "s3://bucket/B", "name": "B",
"lastUpdate": None},
+ ],
+ }
+
+ def test_two_assets_only_one_queued_last_update_norm(self, test_client,
dag_maker):
+ with dag_maker(
+ dag_id="two_assets",
+ schedule=[
+ Asset(uri="s3://bucket/A", name="A"),
+ Asset(uri="s3://bucket/B", name="B"),
+ ],
+ serialized=True,
+ ):
+ EmptyOperator(task_id="t")
+
+ dr = dag_maker.create_dagrun()
+ dag_maker.sync_dagbag_to_db()
+
+ with create_session() as session:
+ assets = {
+ a.uri: a
+ for a in session.query(AssetModel).filter(
+ AssetModel.uri.in_(["s3://bucket/A", "s3://bucket/B"])
+ )
+ }
+ session.add(AssetDagRunQueue(asset_id=assets["s3://bucket/A"].id,
target_dag_id="two_assets"))
+ session.add(
+ AssetEvent(asset_id=assets["s3://bucket/A"].id,
timestamp=dr.logical_date or pendulum.now())
+ )
+ session.commit()
+
+ resp = test_client.get("/next_run_assets/two_assets")
+ assert resp.status_code == 200
+ events = resp.json()["events"]
+ # ordered by uri => A then B
+ assert events[0]["uri"] == "s3://bucket/A"
+ assert events[0]["lastUpdate"] is not None
+ assert "queued" not in events[0]
+ assert events[1]["uri"] == "s3://bucket/B"
+ assert events[1]["lastUpdate"] is None
+ assert "queued" not in events[1]
+
+ def test_last_update_respects_latest_run_filter(self, test_client,
dag_maker):
+ with dag_maker(
+ dag_id="filter_run",
+ schedule=[Asset(uri="s3://bucket/F", name="F")],
+ serialized=True,
+ ):
+ EmptyOperator(task_id="t")
+
+ dr = dag_maker.create_dagrun()
+ dag_maker.sync_dagbag_to_db()
+
+ with create_session() as session:
Review Comment:
session 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]
