Andrushika commented on code in PR #76932:
URL: https://github.com/apache/airflow/pull/76932#discussion_r4547034355
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/event_logs.py:
##########
@@ -228,40 +257,77 @@ def get_event_logs(
if region_index is not None:
attempt_ids = attempt_ids.where(attempts.c.region_index ==
region_index)
query = query.where(Log.task_instance_id.in_(attempt_ids))
+ filters: list[OrmClause] = [
+ # Exact match filters
+ dag_id,
+ task_id,
+ run_id,
+ map_index,
+ try_number,
+ task_instance_id,
+ owner,
+ event,
+ excluded_events,
+ included_events,
+ before,
+ after,
+ # Pattern search filters
+ dag_id_pattern,
+ dag_id_prefix_pattern,
+ task_id_pattern,
+ task_id_prefix_pattern,
+ run_id_pattern,
+ run_id_prefix_pattern,
+ owner_pattern,
+ owner_prefix_pattern,
+ owner_display_name_pattern,
+ owner_display_name_prefix_pattern,
+ event_pattern,
+ event_prefix_pattern,
+ teams,
+ # Permission
+ readable_event_logs_filter,
+ ]
+
+ if cursor is not None:
+ page_limit = cast("int", limit.value)
+ cursor_limit = LimitFilter().set_value(page_limit + 1)
+ event_logs_select = apply_filters_to_select(statement=query,
filters=[*filters, cursor_limit])
+ event_logs_select = order_by.to_orm(event_logs_select)
+
+ is_backward = False
+ if cursor:
+ token, is_backward = parse_cursor(cursor)
+ if is_backward:
+ event_logs_select = order_by.to_orm(event_logs_select,
reversed=True)
+ event_logs_select = apply_cursor_filter(
+ event_logs_select, token, order_by,
session.get_bind().dialect.name, is_backward=is_backward
+ )
+
+ fetched = list(session.scalars(event_logs_select))
+ has_more = len(fetched) > page_limit
+ page = fetched[:page_limit]
+
+ if is_backward:
+ page.reverse()
+ has_prev = has_more
+ has_next = True
+ else:
+ has_prev = bool(cursor)
+ has_next = has_more
+
+ return EventLogCollectionResponse(
+ event_logs=[event_log_to_response(event_log=event_log) for
event_log in page],
+ next_cursor=encode_cursor(page[-1], order_by) if has_next and page
else None,
+ previous_cursor=(
+ make_backward_cursor(encode_cursor(page[0], order_by)) if
has_prev and page else None
+ ),
+ )
Review Comment:
Could we encode the cursors before `event_log_to_response`? It fills an
empty `owner_display_name` from `owner`, and with
`order_by=-owner_display_name` the `next_cursor` loops forever.
```suggestion
next_cursor = encode_cursor(page[-1], order_by) if has_next and page
else None
previous_cursor = (
make_backward_cursor(encode_cursor(page[0], order_by)) if
has_prev and page else None
)
return EventLogCollectionResponse(
event_logs=[event_log_to_response(event_log=event_log) for
event_log in page],
next_cursor=next_cursor,
previous_cursor=previous_cursor,
)
```
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_event_logs.py:
##########
@@ -829,3 +829,52 @@ def test_non_dag_rows_are_gated_on_audit_logs_all(
mock_is_authorized_view.assert_called_once_with(
access_view=AccessView.AUDIT_LOGS_ALL, user=mock.ANY,
team_name=None
)
+
+ @pytest.mark.parametrize("order_by", ["-when", "event", "owner",
"-event_log_id"])
+ def test_cursor_pagination_forward_then_previous_page(self, test_client,
order_by):
+ """Every row is reached once going forward, and previous_cursor
retraces the same pages."""
+ forward_pages = []
+ cursor = ""
+ while cursor is not None:
+ response = test_client.get(
+ "/eventLogs", params={"limit": 3, "order_by": order_by,
"cursor": cursor}
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ forward_pages.append(body)
+ cursor = body["next_cursor"]
+
+ forward_ids = [log["event_log_id"] for page in forward_pages for log
in page["event_logs"]]
+ assert len(forward_pages) == 2
+ assert len(forward_ids) == len(set(forward_ids)) == 4
+ assert forward_pages[0]["previous_cursor"] is None
+
+ response = test_client.get(
+ "/eventLogs",
+ params={"limit": 3, "order_by": order_by, "cursor":
forward_pages[-1]["previous_cursor"]},
+ )
+ assert response.status_code == 200
+ body = response.json()
+ assert [log["event_log_id"] for log in body["event_logs"]] ==
forward_ids[:3]
+ assert body["previous_cursor"] is None
+ assert body["next_cursor"] is not None
+
+ def test_cursor_pagination_does_not_count_rows(self, test_client):
+ with assert_queries_count(2):
+ response = test_client.get("/eventLogs", params={"cursor": ""})
+ assert response.status_code == 200
+ assert response.json()["total_entries"] is None
+
+ def test_cursor_pagination_keeps_filters_on_every_page(self, test_client):
+ events = []
+ cursor = ""
+ while cursor is not None:
+ response = test_client.get(
+ "/eventLogs", params={"limit": 1, "dag_id": DAG_ID,
"order_by": "event", "cursor": cursor}
+ )
+ assert response.status_code == 200, response.json()
+ body = response.json()
+ events.extend(log["event"] for log in body["event_logs"])
+ cursor = body["next_cursor"]
+
+ assert events == [EVENT_WITH_OWNER_AND_TASK_INSTANCE,
TASK_INSTANCE_EVENT]
Review Comment:
This test fails on the current head and passes with the fix above.
```suggestion
assert events == [EVENT_WITH_OWNER_AND_TASK_INSTANCE,
TASK_INSTANCE_EVENT]
@pytest.mark.parametrize("order_by", ["owner_display_name",
"-owner_display_name"])
def test_cursor_pagination_matches_offset_order(self, test_client,
order_by):
expected_ids = [
log["event_log_id"]
for log in test_client.get("/eventLogs", params={"order_by":
order_by}).json()["event_logs"]
]
event_log_ids = []
cursor = ""
for _ in expected_ids:
response = test_client.get(
"/eventLogs", params={"limit": 1, "order_by": order_by,
"cursor": cursor}
)
assert response.status_code == 200, response.json()
body = response.json()
event_log_ids.extend(log["event_log_id"] for log in
body["event_logs"])
cursor = body["next_cursor"]
assert event_log_ids == expected_ids
assert cursor is None
```
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/event_logs.py:
##########
@@ -52,7 +52,21 @@ class EventLogResponse(BaseModel):
class EventLogCollectionResponse(BaseModel):
- """Event Log Collection Response."""
+ """Event Log collection response supporting both offset and cursor
pagination."""
event_logs: Iterable[EventLogResponse]
- total_entries: int
+ total_entries: int | None = Field(
Review Comment:
Would `bounded_total_entries` work here, like the Dag run and task instance
lists after #71252?
##########
airflow-core/src/airflow/ui/src/utils/executionParams.ts:
##########
@@ -22,10 +22,12 @@ const EXECUTION_PARAMS = [
SearchParamsKeys.TRY_NUMBER,
SearchParamsKeys.REGION_ID,
SearchParamsKeys.REGION_INDEX,
+ SearchParamsKeys.CURSOR,
Review Comment:
Could the description mention this? It also changes the task links in Grid,
Graph and Gantt.
--
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]
