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

Code before:
            json_pod = PodGenerator.serialize_pod(var)
            return cls._encode(json_pod, type_=DAT.POD)
        elif isinstance(var, OutletEventAccessors):
            return cls._encode(
                encode_outlet_event_accessors(var),
                type_=DAT.ASSET_EVENT_ACCESSORS,
            )
        elif isinstance(var, AssetUniqueKey):
            return cls._encode(
                attrs.asdict(var),
                type_=DAT.ASSET_UNIQUE_KEY,
            )
        elif isinstance(var, AssetAliasUniqueKey):
            return cls._encode(
                attrs.asdict(var),
                type_=DAT.ASSET_ALIAS_UNIQUE_KEY,
            )
        elif isinstance(var, DAG):
            return cls._encode(DagSerialization.serialize_dag(var), type_=DAT.DAG)
        elif isinstance(var, (DeadlineAlert, SerializedDeadlineAlert)):
            from airflow.serialization.encoders import encode_deadline_alert

            return cls._encode(encode_deadline_alert(var), type_=DAT.DEADLINE_ALERT)
        elif isinstance(var, Resources):
            return var.to_dict()
        elif isinstance(var, MappedOperator):
            return cls._encode(OperatorSerialization.serialize_mapped_operator(var), type_=DAT.OP)
        elif isinstance(var, BaseOperator):
            var._needs_expansion = var.get_needs_expansion()
            return cls._encode(OperatorSerialization.serialize_operator(var), type_=DAT.OP)
        elif isinstance(var, cls._datetime_types):
            return cls._encode(var.timestamp(), type_=DAT.DATETIME)
        elif isinstance(var, datetime.timedelta):
            return cls._encode(var.total_seconds(), type_=DAT.TIMEDELTA)
        elif isinstance(var, (Timezone, FixedTimezone)):
            return cls._encode(encode_timezone(var), type_=DAT.TIMEZONE)
        elif isinstance(var, relativedelta.relativedelta):
            return cls._encode(encode_relativedelta(var), type_=DAT.RELATIVEDELTA)
        elif isinstance(var, TaskInstanceKey):
            return cls._encode(
                var._asdict(),
                type_=DAT.TASK_INSTANCE_KEY,
            )
        elif isinstance(var, (AirflowException, TaskDeferred)) and hasattr(var, "serialize"):
            exc_cls_name, args, kwargs = var.serialize()
            return cls._encode(
                cls.serialize(
                    {"exc_cls_name": exc_cls_name, "args": args, "kwargs": kwargs},
                    strict=strict,
                ),
                type_=DAT.AIRFLOW_EXC_SER,

Review comment (suggestion):
I'm not sure I follow why it can be either one.  If you are attempting to serialize then shouldn't it be  the unserialized version coming in?  Or is this just defensive in case we try to re-serialize?  I don't see this pattern in the other elif clauses so it seems a odd enough to ask.

Code after:
                {str(k): cls.serialize(v, strict=strict) for k, v in var.items()},
                type_=DAT.DICT,
            )
        elif isinstance(var, list):
            return [cls.serialize(v, strict=strict) for v in var]
        elif var.__class__.__name__ == "V1Pod" and _has_kubernetes() and isinstance(var, k8s.V1Pod):
            json_pod = PodGenerator.serialize_pod(var)
            return cls._encode(json_pod, type_=DAT.POD)
        elif isinstance(var, OutletEventAccessors):
            return cls._encode(
                encode_outlet_event_accessors(var),
                type_=DAT.ASSET_EVENT_ACCESSORS,
            )
        elif isinstance(var, AssetUniqueKey):
            return cls._encode(
                attrs.asdict(var),
                type_=DAT.ASSET_UNIQUE_KEY,
            )
        elif isinstance(var, AssetAliasUniqueKey):
            return cls._encode(
                attrs.asdict(var),
                type_=DAT.ASSET_ALIAS_UNIQUE_KEY,
            )
        elif isinstance(var, DAG):
            return cls._encode(DagSerialization.serialize_dag(var), type_=DAT.DAG)
        elif isinstance(var, (DeadlineAlert, SerializedDeadlineAlert)):
            return cls._encode(encode_deadline_alert(var), type_=DAT.DEADLINE_ALERT)
        elif isinstance(var, Resources):
            return var.to_dict()
        elif isinstance(var, MappedOperator):
            return cls._encode(OperatorSerialization.serialize_mapped_operator(var), type_=DAT.OP)
        elif isinstance(var, BaseOperator):
            var._needs_expansion = var.get_needs_expansion()
            return cls._encode(OperatorSerialization.serialize_operator(var), type_=DAT.OP)
        elif isinstance(var, cls._datetime_types):
            return cls._encode(var.timestamp(), type_=DAT.DATETIME)
        elif isinstance(var, datetime.timedelta):
            return cls._encode(var.total_seconds(), type_=DAT.TIMEDELTA)
        elif isinstance(var, (Timezone, FixedTimezone)):
            return cls._encode(encode_timezone(var), type_=DAT.TIMEZONE)
        elif isinstance(var, relativedelta.relativedelta):
            return cls._encode(encode_relativedelta(var), type_=DAT.RELATIVEDELTA)
        elif isinstance(var, TaskInstanceKey):
            return cls._encode(
                var._asdict(),
                type_=DAT.TASK_INSTANCE_KEY,
            )
        elif isinstance(var, (AirflowException, TaskDeferred)) and hasattr(var, "serialize"):
            exc_cls_name, args, kwargs = var.serialize()
            return cls._encode(
                cls.serialize(