Repository: apache/airflow
File: airflow-core/src/airflow/serialization/serialized_objects.py (Python)

Code before:

        serialized_op["_is_mapped"] = True
        return serialized_op

    @classmethod
    def serialize_operator(cls, op: SdkOperator) -> dict[str, Any]:
        return cls._serialize_node(op)

    @classmethod
    def _serialize_node(cls, op: SdkOperator) -> dict[str, Any]:
        """Serialize operator into a JSON object."""
        serialize_op = cls.serialize_to_json(op, cls._decorated_fields)

        if not op.email:
            # If "email" is empty, we do not need to include other email attrs
            for attr in ["email_on_failure", "email_on_retry"]:
                if attr in serialize_op:
                    del serialize_op[attr]

        # Detect if there's a change in python callable name
        python_callable = getattr(op, "python_callable", None)
        if python_callable:
            callable_name = qualname(python_callable)
            serialize_op["python_callable_name"] = callable_name

        serialize_op["task_type"] = getattr(op, "task_type", type(op).__name__)
        serialize_op["_task_module"] = getattr(op, "_task_module", type(op).__module__)
        if op.operator_name != serialize_op["task_type"]:
            serialize_op["_operator_name"] = op.operator_name

        # Used to determine if an Operator is inherited from EmptyOperator
        if op.inherits_from_empty_operator:
            serialize_op["_is_empty"] = True

        # Used to determine if an Operator is inherited from SkipMixin or BranchMixin
        if op.inherits_from_skipmixin:
            serialize_op["_can_skip_downstream"] = True

        if op.start_trigger_args:
            serialize_op["start_trigger_args"] = encode_start_trigger_args(op.start_trigger_args)

        if op.operator_extra_links:
            serialize_op["_operator_extra_links"] = cls._serialize_operator_extra_links(
                op.operator_extra_links.__get__(op)
                if isinstance(op.operator_extra_links, property)
                else op.operator_extra_links
            )

        # Store all template_fields as they are if there are JSON Serializable
        # If not, store them as strings
        # And raise an exception if the field is not templateable

Review comment (suggestion):
Was just reviewing in consideration to collision between tasks but since its scoped deep in the nested structure, we are good

Code after:
            )
            del serialized_op["partial_kwargs"]["python_callable"]

        serialized_op["_is_mapped"] = True
        return serialized_op

    @classmethod
    def serialize_operator(cls, op: SdkOperator) -> dict[str, Any]:
        return cls._serialize_node(op)

    @classmethod
    def _serialize_node(cls, op: SdkOperator) -> dict[str, Any]:
        """Serialize operator into a JSON object."""
        serialize_op = cls.serialize_to_json(op, cls._decorated_fields)

        if not op.email:
            # If "email" is empty, we do not need to include other email attrs
            for attr in ["email_on_failure", "email_on_retry"]:
                if attr in serialize_op:
                    del serialize_op[attr]

        # Store python_callable_name for change detection.
        # exclude_module=True ensures stable names across bundle version changes.
        python_callable = getattr(op, "python_callable", None)
        if python_callable:
            serialize_op["python_callable_name"] = qualname(python_callable, exclude_module=True)

        serialize_op["task_type"] = getattr(op, "task_type", type(op).__name__)
        serialize_op["_task_module"] = getattr(op, "_task_module", type(op).__module__)
        if op.operator_name != serialize_op["task_type"]:
            serialize_op["_operator_name"] = op.operator_name

        # Used to determine if an Operator is inherited from EmptyOperator
        if op.inherits_from_empty_operator:
            serialize_op["_is_empty"] = True

        # Used to determine if an Operator is inherited from SkipMixin or BranchMixin
        if op.inherits_from_skipmixin:
            serialize_op["_can_skip_downstream"] = True

        if op.start_trigger_args:
            serialize_op["start_trigger_args"] = encode_start_trigger_args(op.start_trigger_args)

        if op.operator_extra_links:
            serialize_op["_operator_extra_links"] = cls._serialize_operator_extra_links(
                op.operator_extra_links.__get__(op)
                if isinstance(op.operator_extra_links, property)
                else op.operator_extra_links
            )

        # Store all template_fields as they are if there are JSON Serializable