ashb commented on code in PR #52234:
URL: https://github.com/apache/airflow/pull/52234#discussion_r2192051819
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py:
##########
@@ -110,7 +109,7 @@ def _fill_task_group_map(
_fill_task_group_map(task_node=child, parent_node=task_node)
return
- if isinstance(task_node, BaseOperator):
+ if isinstance(task_node, (BaseOperator, SerializedBaseOperator)):
Review Comment:
This can't ever be a `BaseOperator` can it?
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/grid.py:
##########
@@ -344,8 +345,8 @@ def _get_aggs_for_node(detail):
def _find_aggregates(
- node: TaskGroup | BaseOperator | MappedTaskGroup | TaskMap,
- parent_node: TaskGroup | BaseOperator | MappedTaskGroup | TaskMap | None,
+ node: TaskGroup | BaseOperator | MappedTaskGroup | SerializedBaseOperator
| TaskMap,
+ parent_node: TaskGroup | BaseOperator | MappedTaskGroup |
SerializedBaseOperator | TaskMap | None,
Review Comment:
Ditto, it's not possible for these to be `BaseOperator`?
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -738,7 +752,7 @@ def serialize(
return var.to_dict()
elif isinstance(var, MappedOperator):
return
cls._encode(SerializedBaseOperator.serialize_mapped_operator(var), type_=DAT.OP)
- elif isinstance(var, TaskSDKBaseOperator):
+ elif isinstance(var, (BaseOperator, SerializedBaseOperator)):
Review Comment:
I don't think it makes sense to serialized an already Serialized operator,
does it?
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -1281,11 +1357,11 @@ def serialize_mapped_operator(cls, op: MappedOperator)
-> dict[str, Any]:
return serialized_op
@classmethod
- def serialize_operator(cls, op: TaskSDKBaseOperator | MappedOperator) ->
dict[str, Any]:
+ def serialize_operator(cls, op: BaseOperator | MappedOperator |
SerializedBaseOperator) -> dict[str, Any]:
Review Comment:
Ditto.
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -881,7 +886,11 @@ def refresh_from_db(
else:
self.state = None
- def refresh_from_task(self, task: Operator, pool_override: str | None =
None) -> None:
+ def refresh_from_task(
+ self,
+ task: Operator | SerializedBaseOperator,
Review Comment:
And here -- when is this used with anything other than a serialized op?
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -116,6 +122,14 @@
except ImportError:
pass
+DEFAULT_OPERATOR_DEPS = {
Review Comment:
Nit: frozenset?
--
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]
