You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "yimingpeng (via GitHub)" <gi...@apache.org> on 2025/11/09 03:51:28 UTC
[PR] Add task group ID filtering support to task instance query [airflow]
yimingpeng opened a new pull request, #58092:
URL: https://github.com/apache/airflow/pull/58092
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
closes: #55319
This PR is to address the 2nd issue of the above ticket.
## Problem summary
When querying with task group name, the task instance list return no results.
## Proposed Solution
Added `QueryTITaskDisplayNameOrGroupPattern` filter in `parameters.py` that now supports
1. The original search pattern
2. An expansion for handling task group, when the search value matches a `task group`, we now query all task IDs within the group.
## After Fix
<img width="1458" height="983" alt="image" src="https://github.com/user-attachments/assets/cc26e00e-af8d-4ec1-8a48-a0e63e53a73a" />
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2603140969
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,64 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskGroupDisplayNamePattern(BaseParam[str]):
+ """Task group display name pattern filter - returns all tasks in matching groups."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+
+ # Pattern matching on both group display name and group_id
+ matching_task_ids = []
+ for group_id, task_group in task_groups.items():
+ if group_id is None: # Skip root group
+ continue
+
+ # Check both the display name (label) and the group_id for pattern matching
+ display_name = getattr(task_group, "label", None)
+ if (
+ display_name and self._matches_pattern(display_name, self.value)
+ ) or self._matches_pattern(group_id, self.value):
+ matching_task_ids.extend([task.task_id for task in task_group.iter_tasks()])
+
+ if matching_task_ids:
+ return select.where(TaskInstance.task_id.in_(matching_task_ids))
Review Comment:
We shouldn't do 'pattern' matching, but exact match on group because it will be too hard to implement properly. What we want are really 'all tasks' from 'this group'. We don't want, 'all tasks' that belongs to a group that match 'that pattern'. (And potentially having cross group TI in the response)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #58092:
URL: https://github.com/apache/airflow/pull/58092#issuecomment-3660611501
Manual backport https://github.com/apache/airflow/pull/59511
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "yimingpeng (via GitHub)" <gi...@apache.org>.
yimingpeng commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2613324584
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -61,6 +61,7 @@
from airflow.models.taskinstance import TaskInstance
from airflow.models.variable import Variable
from airflow.models.xcom import XComModel
+from airflow.serialization.serialized_objects import SerializedDAG
Review Comment:
oops, sorry for overlooking this, fixed.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2603117761
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,64 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskGroupDisplayNamePattern(BaseParam[str]):
+ """Task group display name pattern filter - returns all tasks in matching groups."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+
+ # Pattern matching on both group display name and group_id
+ matching_task_ids = []
+ for group_id, task_group in task_groups.items():
+ if group_id is None: # Skip root group
+ continue
+
+ # Check both the display name (label) and the group_id for pattern matching
+ display_name = getattr(task_group, "label", None)
+ if (
+ display_name and self._matches_pattern(display_name, self.value)
+ ) or self._matches_pattern(group_id, self.value):
+ matching_task_ids.extend([task.task_id for task in task_group.iter_tasks()])
+
+ if matching_task_ids:
+ return select.where(TaskInstance.task_id.in_(matching_task_ids))
+
+ return select.where(TaskInstance.task_id.is_(None))
+
+ def _matches_pattern(self, display_name: str, pattern: str) -> bool:
+ """Check if display_name matches the SQL LIKE pattern or exact match."""
+ import re
+
+ if "%" in pattern:
+ pattern_temp = pattern.replace("%", "\x00").replace("_", "\x01")
+ escaped = re.escape(pattern_temp)
+ regex_pattern = escaped.replace("\x00", ".*").replace("\x01", ".")
+ return bool(re.match(f"^{regex_pattern}$", display_name, re.IGNORECASE))
+ return display_name.lower() == pattern.lower()
Review Comment:
Maybe to keep things simple, we should just do an 'exact' match on the group.
It's not really a 'group' pattern, because we are not returning all groups that match a pattern.
We are returning all TIs that matche a precise group. So a plain `QueryTaskInstanceTaskGroupFilter` is enough.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "yimingpeng (via GitHub)" <gi...@apache.org>.
yimingpeng commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2613324584
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -61,6 +61,7 @@
from airflow.models.taskinstance import TaskInstance
from airflow.models.variable import Variable
from airflow.models.xcom import XComModel
+from airflow.serialization.serialized_objects import SerializedDAG
Review Comment:
oops, sorry for overlooking this, thanks for picking it up, fixed now
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2516684434
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -184,6 +184,44 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskDisplayNameOrGroupPattern(_SearchParam):
+ """Task display name search with task group expansion."""
+
+ def __init__(self, attribute: ColumnElement, dag=None, skip_none: bool = True):
+ super().__init__(attribute, skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ conditions = [self.attribute.ilike(f"%{self.value}%")]
+
+ # Task group expansion: check if value matches a task group ID
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+ if target_group := task_groups.get(self.value):
+ task_ids = [task.task_id for task in target_group.iter_tasks()]
+ conditions.append(TaskInstance.task_id.in_(task_ids))
+
+ return select.where(or_(*conditions))
+
+ @classmethod
+ def depends(
+ cls,
+ value: str | None = Query(
+ alias="task_display_name_pattern",
+ default=None,
+ description="SQL LIKE expression to filter by task display name — use `%` / `_` wildcards "
+ "(e.g. `%customer_%`). Regular expressions are **not** supported. "
+ "Can also be a task group ID to filter by all tasks within that group.",
Review Comment:
I think this should be task group display name, which is different from task group id
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2592427789
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,64 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskGroupDisplayNamePattern(BaseParam[str]):
+ """Task group display name pattern filter - returns all tasks in matching groups."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+
+ # Pattern matching on both group display name and group_id
+ matching_task_ids = []
+ for group_id, task_group in task_groups.items():
+ if group_id is None: # Skip root group
+ continue
+
+ # Check both the display name (label) and the group_id for pattern matching
+ display_name = getattr(task_group, "label", None)
+ if (
+ display_name and self._matches_pattern(display_name, self.value)
+ ) or self._matches_pattern(group_id, self.value):
+ matching_task_ids.extend([task.task_id for task in task_group.iter_tasks()])
+
+ if matching_task_ids:
+ return select.where(TaskInstance.task_id.in_(matching_task_ids))
Review Comment:
The pattern search will be used with `LIKE` in SQL statement for the other entities.
https://github.com/apache/airflow/blob/52526b3f6051357587d4826e993e7fe4515b7876/airflow-core/src/airflow/api_fastapi/common/parameters.py#L173-L176
However, in this case, we are only able to get `SerializedDAG` and not able to use SQL to filter on those name with `LIKE`. Would it be better to use `FilterOptionEnum.IN` and use `if self.value in display_name` be more appropriate?
cc @pierrejeambrun
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,64 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskGroupDisplayNamePattern(BaseParam[str]):
+ """Task group display name pattern filter - returns all tasks in matching groups."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self.dag = dag
Review Comment:
Would it be better to provide a setter for `dag` attribute and add validation at `to_orm` to check whether the `dag` attribute is set? Just in case we need the same query class in some other place and having a explicit setter might be more readable.
```suggestion
self.dag: None | SerializedDAG = dag
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2517532839
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -184,6 +184,44 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskDisplayNameOrGroupPattern(_SearchParam):
+ """Task display name search with task group expansion."""
+
+ def __init__(self, attribute: ColumnElement, dag=None, skip_none: bool = True):
+ super().__init__(attribute, skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ conditions = [self.attribute.ilike(f"%{self.value}%")]
+
+ # Task group expansion: check if value matches a task group ID
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+ if target_group := task_groups.get(self.value):
Review Comment:
If we are searching by pattern, should we get all key contain the pattern instead of directly get by value?
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -184,6 +184,44 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskDisplayNameOrGroupPattern(_SearchParam):
+ """Task display name search with task group expansion."""
+
+ def __init__(self, attribute: ColumnElement, dag=None, skip_none: bool = True):
+ super().__init__(attribute, skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ conditions = [self.attribute.ilike(f"%{self.value}%")]
+
+ # Task group expansion: check if value matches a task group ID
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+ if target_group := task_groups.get(self.value):
+ task_ids = [task.task_id for task in target_group.iter_tasks()]
+ conditions.append(TaskInstance.task_id.in_(task_ids))
+
+ return select.where(or_(*conditions))
+
+ @classmethod
+ def depends(
+ cls,
+ value: str | None = Query(
+ alias="task_display_name_pattern",
+ default=None,
+ description="SQL LIKE expression to filter by task display name — use `%` / `_` wildcards "
+ "(e.g. `%customer_%`). Regular expressions are **not** supported. "
+ "Can also be a task group ID to filter by all tasks within that group.",
Review Comment:
Or maybe we could have another filter just for task group?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2610557255
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,53 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTaskInstanceTaskGroupFilter(BaseParam[str]):
+ """Task group filter - returns all tasks in the specified group."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self._dag: None | Any = dag
+
+ @property
+ def dag(self) -> None | Any:
+ return self._dag
+
+ @dag.setter
+ def dag(self, value: None | Any) -> None:
+ self._dag = value
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if not self.dag:
+ raise ValueError("DAG must be set before calling to_orm")
+
+ if not hasattr(self.dag, "task_group"):
+ return select
+
+ matching_task_ids = []
+
+ # Exact matching on group_id
+ task_groups = self.dag.task_group.get_task_group_dict()
+ task_group = task_groups.get(self.value)
+ if task_group:
+ matching_task_ids = [task.task_id for task in task_group.iter_tasks()]
+
+ return select.where(TaskInstance.task_id.in_(matching_task_ids))
Review Comment:
It seems we could simplify the filter as:
```suggestion
return select.where(TaskInstance.task_id.in_(task.task_id for task in task_group.iter_tasks()))
```
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,53 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTaskInstanceTaskGroupFilter(BaseParam[str]):
+ """Task group filter - returns all tasks in the specified group."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self._dag: None | Any = dag
+
+ @property
+ def dag(self) -> None | Any:
+ return self._dag
+
+ @dag.setter
+ def dag(self, value: None | Any) -> None:
+ self._dag = value
Review Comment:
The `SerializedDAG` can be imported from `airflow.serialization.serialized_objects` under `TYPE_CHECKING` block.
```suggestion
self._dag: None | SerializedDAG = dag
@property
def dag(self) -> None | SerializedDAG:
return self._dag
@dag.setter
def dag(self, value: SerializedDAG) -> None:
self._dag = value
```
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,53 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTaskInstanceTaskGroupFilter(BaseParam[str]):
+ """Task group filter - returns all tasks in the specified group."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self._dag: None | Any = dag
+
+ @property
+ def dag(self) -> None | Any:
+ return self._dag
+
+ @dag.setter
+ def dag(self, value: None | Any) -> None:
+ self._dag = value
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if not self.dag:
+ raise ValueError("DAG must be set before calling to_orm")
Review Comment:
```suggestion
raise ValueError("Dag must be set before calling to_orm")
```
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,53 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTaskInstanceTaskGroupFilter(BaseParam[str]):
+ """Task group filter - returns all tasks in the specified group."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self._dag: None | Any = dag
+
+ @property
+ def dag(self) -> None | Any:
+ return self._dag
+
+ @dag.setter
+ def dag(self, value: None | Any) -> None:
+ self._dag = value
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if not self.dag:
+ raise ValueError("DAG must be set before calling to_orm")
+
+ if not hasattr(self.dag, "task_group"):
+ return select
+
+ matching_task_ids = []
+
+ # Exact matching on group_id
Review Comment:
Not sure should we raise Not Found error if the task_group not found like:
https://github.com/apache/airflow/blob/7c6fcecc535de447482b18c85fb98235655e745d/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py#L950-L976
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun merged PR #58092:
URL: https://github.com/apache/airflow/pull/58092
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #58092:
URL: https://github.com/apache/airflow/pull/58092#issuecomment-3596714763
@yimingpeng That would be useful for other PRs/features too such as https://github.com/apache/airflow/pull/58102. Any chance we can address the comment so we can move this forward?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "yimingpeng (via GitHub)" <gi...@apache.org>.
yimingpeng commented on PR #58092:
URL: https://github.com/apache/airflow/pull/58092#issuecomment-3615850786
Hello @guan404ming @jason810496 @pierrejeambrun ,
Thanks for your review comments and apologise for the delay.
I think that I addressed all your feedbacks, pls help have another review when you have a chance, many thanks.
https://github.com/user-attachments/assets/ccd11a19-0637-4a73-b2b3-7ebbbbf74395
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "yimingpeng (via GitHub)" <gi...@apache.org>.
yimingpeng commented on PR #58092:
URL: https://github.com/apache/airflow/pull/58092#issuecomment-3640732283
Thanks for insightful feedback, @jason810496 @pierrejeambrun
Based on what has been suggested, I've made another update, please have another review when you have a chance, thanks 🙏.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2603125817
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,64 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskGroupDisplayNamePattern(BaseParam[str]):
+ """Task group display name pattern filter - returns all tasks in matching groups."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+
+ # Pattern matching on both group display name and group_id
+ matching_task_ids = []
+ for group_id, task_group in task_groups.items():
+ if group_id is None: # Skip root group
+ continue
+
+ # Check both the display name (label) and the group_id for pattern matching
+ display_name = getattr(task_group, "label", None)
+ if (
+ display_name and self._matches_pattern(display_name, self.value)
+ ) or self._matches_pattern(group_id, self.value):
+ matching_task_ids.extend([task.task_id for task in task_group.iter_tasks()])
+
+ if matching_task_ids:
+ return select.where(TaskInstance.task_id.in_(matching_task_ids))
+
+ return select.where(TaskInstance.task_id.is_(None))
Review Comment:
Why that? Maybe just always return `select.where(TaskInstance.task_id.in_(matching_task_ids))`. If matching_task_ids is empty [], no task will match the filter, which is expected.
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,64 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskGroupDisplayNamePattern(BaseParam[str]):
+ """Task group display name pattern filter - returns all tasks in matching groups."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+
+ # Pattern matching on both group display name and group_id
+ matching_task_ids = []
+ for group_id, task_group in task_groups.items():
+ if group_id is None: # Skip root group
+ continue
+
+ # Check both the display name (label) and the group_id for pattern matching
+ display_name = getattr(task_group, "label", None)
+ if (
+ display_name and self._matches_pattern(display_name, self.value)
+ ) or self._matches_pattern(group_id, self.value):
+ matching_task_ids.extend([task.task_id for task in task_group.iter_tasks()])
+
+ if matching_task_ids:
+ return select.where(TaskInstance.task_id.in_(matching_task_ids))
+
+ return select.where(TaskInstance.task_id.is_(None))
+
+ def _matches_pattern(self, display_name: str, pattern: str) -> bool:
+ """Check if display_name matches the SQL LIKE pattern or exact match."""
+ import re
+
+ if "%" in pattern:
+ pattern_temp = pattern.replace("%", "\x00").replace("_", "\x01")
+ escaped = re.escape(pattern_temp)
+ regex_pattern = escaped.replace("\x00", ".*").replace("\x01", ".")
+ return bool(re.match(f"^{regex_pattern}$", display_name, re.IGNORECASE))
+ return display_name.lower() == pattern.lower()
Review Comment:
Maybe to keep things simple, we should just do an 'exact' match on the group.
It's not really a 'group' pattern, because we are not returning all groups that match a pattern.
We are returning all TI that matches a precise group. So a plain `QueryTaskInstanceTaskGroupFilter` is enough.
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,64 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskGroupDisplayNamePattern(BaseParam[str]):
+ """Task group display name pattern filter - returns all tasks in matching groups."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+
+ # Pattern matching on both group display name and group_id
+ matching_task_ids = []
+ for group_id, task_group in task_groups.items():
+ if group_id is None: # Skip root group
+ continue
+
+ # Check both the display name (label) and the group_id for pattern matching
+ display_name = getattr(task_group, "label", None)
+ if (
+ display_name and self._matches_pattern(display_name, self.value)
+ ) or self._matches_pattern(group_id, self.value):
+ matching_task_ids.extend([task.task_id for task in task_group.iter_tasks()])
+
+ if matching_task_ids:
+ return select.where(TaskInstance.task_id.in_(matching_task_ids))
+
+ return select.where(TaskInstance.task_id.is_(None))
+
+ def _matches_pattern(self, display_name: str, pattern: str) -> bool:
+ """Check if display_name matches the SQL LIKE pattern or exact match."""
+ import re
+
+ if "%" in pattern:
+ pattern_temp = pattern.replace("%", "\x00").replace("_", "\x01")
+ escaped = re.escape(pattern_temp)
+ regex_pattern = escaped.replace("\x00", ".*").replace("\x01", ".")
+ return bool(re.match(f"^{regex_pattern}$", display_name, re.IGNORECASE))
+ return display_name.lower() == pattern.lower()
Review Comment:
We shouldn't do regexp matching on untrusted user provided values.
This is a vector attack for ReDOS attack.
Also it will be impossible to support all the special characters. For instance `_` etc...
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2613219604
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -61,6 +61,7 @@
from airflow.models.taskinstance import TaskInstance
from airflow.models.variable import Variable
from airflow.models.xcom import XComModel
+from airflow.serialization.serialized_objects import SerializedDAG
Review Comment:
The import of `SerializedDAG` needs to be moved under `TYPE_CHECKING` block, as we only use `SerializedDAG` for type annotation.
```python
if TYPE_CHECKING:
from sqlalchemy.orm.attributes import InstrumentedAttribute
from sqlalchemy.sql import ColumnElement, Select
from airflow.serialization.serialized_objects import SerializedDAG
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "yimingpeng (via GitHub)" <gi...@apache.org>.
yimingpeng commented on PR #58092:
URL: https://github.com/apache/airflow/pull/58092#issuecomment-3645208941
Hi @jason810496, thanks for the quick review and the good suggestions, just updated accordingly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2603117761
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -185,6 +185,64 @@ def depends(cls, *args: Any, **kwargs: Any) -> Self:
raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+class QueryTITaskGroupDisplayNamePattern(BaseParam[str]):
+ """Task group display name pattern filter - returns all tasks in matching groups."""
+
+ def __init__(self, dag=None, skip_none: bool = True):
+ super().__init__(skip_none=skip_none)
+ self.dag = dag
+
+ def to_orm(self, select: Select) -> Select:
+ if self.value is None and self.skip_none:
+ return select
+
+ if self.dag and hasattr(self.dag, "task_group"):
+ task_groups = self.dag.task_group.get_task_group_dict()
+
+ # Pattern matching on both group display name and group_id
+ matching_task_ids = []
+ for group_id, task_group in task_groups.items():
+ if group_id is None: # Skip root group
+ continue
+
+ # Check both the display name (label) and the group_id for pattern matching
+ display_name = getattr(task_group, "label", None)
+ if (
+ display_name and self._matches_pattern(display_name, self.value)
+ ) or self._matches_pattern(group_id, self.value):
+ matching_task_ids.extend([task.task_id for task in task_group.iter_tasks()])
+
+ if matching_task_ids:
+ return select.where(TaskInstance.task_id.in_(matching_task_ids))
+
+ return select.where(TaskInstance.task_id.is_(None))
+
+ def _matches_pattern(self, display_name: str, pattern: str) -> bool:
+ """Check if display_name matches the SQL LIKE pattern or exact match."""
+ import re
+
+ if "%" in pattern:
+ pattern_temp = pattern.replace("%", "\x00").replace("_", "\x01")
+ escaped = re.escape(pattern_temp)
+ regex_pattern = escaped.replace("\x00", ".*").replace("\x01", ".")
+ return bool(re.match(f"^{regex_pattern}$", display_name, re.IGNORECASE))
+ return display_name.lower() == pattern.lower()
Review Comment:
Maybe to keep things simple, we should just do an 'exact' match on the group.
It's not really a 'group' pattern, because we are not returning all groups that match a pattern.
We are returning all TIs that matche a precise group. So a plain `QueryTaskInstanceTaskGroupFilter` is enough I think and will make it simpler to implement.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2614138740
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -424,6 +425,7 @@ def get_task_instances(
update_at_range: Annotated[RangeFilter, Depends(datetime_range_filter_factory("updated_at", TI))],
duration_range: Annotated[RangeFilter, Depends(float_range_filter_factory("duration", TI))],
task_display_name_pattern: QueryTITaskDisplayNamePatternSearch,
+ task_group: QueryTITaskGroupFilter,
Review Comment:
Sorry for the nitpicking, since this is public API we need to make sure we are consistent and things respect the naming convention because we will not be able to update them afterward.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -424,6 +425,7 @@ def get_task_instances(
update_at_range: Annotated[RangeFilter, Depends(datetime_range_filter_factory("updated_at", TI))],
duration_range: Annotated[RangeFilter, Depends(float_range_filter_factory("duration", TI))],
task_display_name_pattern: QueryTITaskDisplayNamePatternSearch,
+ task_group: QueryTITaskGroupFilter,
Review Comment:
```suggestion
task_group_id: QueryTITaskGroupFilter,
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -424,6 +425,7 @@ def get_task_instances(
update_at_range: Annotated[RangeFilter, Depends(datetime_range_filter_factory("updated_at", TI))],
duration_range: Annotated[RangeFilter, Depends(float_range_filter_factory("duration", TI))],
task_display_name_pattern: QueryTITaskDisplayNamePatternSearch,
+ task_group: QueryTITaskGroupFilter,
Review Comment:
```suggestion
task_group_id: QueryTITaskGroupFilter,
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -424,6 +425,7 @@ def get_task_instances(
update_at_range: Annotated[RangeFilter, Depends(datetime_range_filter_factory("updated_at", TI))],
duration_range: Annotated[RangeFilter, Depends(float_range_filter_factory("duration", TI))],
task_display_name_pattern: QueryTITaskDisplayNamePatternSearch,
+ task_group: QueryTITaskGroupFilter,
Review Comment:
Sorry for the nitpicking, since this is public API we need to make sure we are consistent and things respect the naming convention because we will not be able to update them afterward.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #58092:
URL: https://github.com/apache/airflow/pull/58092#issuecomment-3647081745
### Backport failed to create: v3-1-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/20171908778'> Run details </a>
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>v3-1-test</td>
<td><a href="https://github.com/apache/airflow/commit/8256ee16d1de5b47ad996af82608d2bb836dfe1d"><img src='https://img.shields.io/badge/Commit-8256ee1-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker 8256ee1 v3-1-test
```
This should apply the commit to the v3-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #58092:
URL: https://github.com/apache/airflow/pull/58092#discussion_r2613567788
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -61,6 +61,7 @@
from airflow.models.taskinstance import TaskInstance
from airflow.models.variable import Variable
from airflow.models.xcom import XComModel
+from airflow.serialization.serialized_objects import SerializedDAG
Review Comment:
Thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task group ID filtering support to task instance query [airflow]
Posted by "yimingpeng (via GitHub)" <gi...@apache.org>.
yimingpeng commented on PR #58092:
URL: https://github.com/apache/airflow/pull/58092#issuecomment-3600925708
Hi @pierrejeambrun, apologies for the delay, just a bit swamped with some other things. I’ll get this sorted within the week and thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org