Nataneljpwd commented on code in PR #53450:
URL: https://github.com/apache/airflow/pull/53450#discussion_r2213382875
##########
airflow-core/src/airflow/utils/__init__.py:
##########
@@ -46,5 +46,11 @@ def __getattr__(name: str):
"xcom": {
"XCOM_RETURN_KEY": "airflow.models.xcom.XCOM_RETURN_KEY",
},
+ "task_group": {
Review Comment:
Why does it have an underscore here and not in the import? It seems
inconsistent, I think it is best to keep the underscore (make the import also
be task_group) instead
##########
task-sdk/src/airflow/sdk/definitions/taskgroup.py:
##########
@@ -715,5 +723,44 @@ def task_group_to_dict(task_item_or_group):
"tooltip": task_group.tooltip,
"is_mapped": is_mapped,
"children": children,
- "type": "task_group",
+ "type": "task",
Review Comment:
Why did you change the type to task? It seems to be a task group
##########
task-sdk/src/airflow/sdk/definitions/taskgroup.py:
##########
@@ -715,5 +723,44 @@ def task_group_to_dict(task_item_or_group):
"tooltip": task_group.tooltip,
"is_mapped": is_mapped,
"children": children,
- "type": "task_group",
+ "type": "task",
+ }
+
+
+def task_group_to_dict_grid(task_item_or_group, parent_group_is_mapped=False):
Review Comment:
What is the difference between this function and the non grid one? They seem
to do the same thing just with a slightly different schema, maybe we only need
1 method that returns the wider schema and the users will use only the fields
they need?
##########
task-sdk/src/airflow/sdk/definitions/taskgroup.py:
##########
@@ -669,36 +672,41 @@ def iter_mapped_dependencies(self) -> Iterator[Operator]:
yield op
-def task_group_to_dict(task_item_or_group):
+@cache
Review Comment:
Won't the cache cause probelms when dags are reserialized? There is nothing
that invalidates the cache other than the lru alg, did I not understand it
correctly?
##########
providers/standard/tests/unit/standard/decorators/test_python.py:
##########
@@ -45,7 +46,13 @@
from airflow.models.expandinput import DictOfListsExpandInput
from airflow.models.mappedoperator import MappedOperator
from airflow.models.xcom_arg import XComArg
- from airflow.utils.task_group import TaskGroup
+ from airflow.utils.task_group import TaskGroup # type: ignore[no-redef]
Review Comment:
I think this import is redundent
##########
task-sdk/src/airflow/sdk/definitions/taskgroup.py:
##########
@@ -715,5 +723,44 @@ def task_group_to_dict(task_item_or_group):
"tooltip": task_group.tooltip,
"is_mapped": is_mapped,
"children": children,
- "type": "task_group",
+ "type": "task",
+ }
+
+
+def task_group_to_dict_grid(task_item_or_group, parent_group_is_mapped=False):
+ """Create a nested dict representation of this TaskGroup and its children
used to construct the Graph."""
+ from airflow.sdk.definitions._internal.abstractoperator import
AbstractOperator
+ from airflow.sdk.definitions.mappedoperator import MappedOperator
+ from airflow.serialization.serialized_objects import SerializedBaseOperator
+
+ if isinstance(task := task_item_or_group, (AbstractOperator,
SerializedBaseOperator)):
+ is_mapped = None
+ if isinstance(task, MappedOperator) or parent_group_is_mapped:
+ is_mapped = True
+ setup_teardown_type = None
+ if task.is_setup is True:
+ setup_teardown_type = "setup"
+ elif task.is_teardown is True:
+ setup_teardown_type = "teardown"
+ return {
+ "id": task.task_id,
+ "label": task.label,
+ "is_mapped": is_mapped,
+ "children": None,
+ "setup_teardown_type": setup_teardown_type,
+ }
+
+ task_group = task_item_or_group
+ task_group_sort = get_task_group_children_getter()
Review Comment:
The name does not reflect that this object gets task group's children,
something like get_sorted_child_tasks should be better
--
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]
