=== Airflow 3.2.2 BaseSerialization.deserialize === 2026-07-07T22:41:47.694326Z [info ] setup plugin alembic.autogenerate.schemas [alembic.runtime.plugins] loc=plugins.py:37 2026-07-07T22:41:47.694487Z [info ] setup plugin alembic.autogenerate.tables [alembic.runtime.plugins] loc=plugins.py:37 2026-07-07T22:41:47.694553Z [info ] setup plugin alembic.autogenerate.types [alembic.runtime.plugins] loc=plugins.py:37 2026-07-07T22:41:47.694608Z [info ] setup plugin alembic.autogenerate.constraints [alembic.runtime.plugins] loc=plugins.py:37 2026-07-07T22:41:47.694675Z [info ] setup plugin alembic.autogenerate.defaults [alembic.runtime.plugins] loc=plugins.py:37 2026-07-07T22:41:47.694743Z [info ] setup plugin alembic.autogenerate.comments [alembic.runtime.plugins] loc=plugins.py:37 version 3.2.2 file /data/pruva/project-cache/03610ec6-e6fb-4086-9681-103dd9199da6/repo/airflow_product_venv/lib/python3.14/site-packages/airflow/serialization/serialized_objects.py line 619 @classmethod def deserialize(cls, encoded_var: Any) -> Any: """ Deserialize an object; helper function of depth first search for deserialization. :meta private: """ if cls._is_primitive(encoded_var): return encoded_var elif isinstance(encoded_var, list): return [cls.deserialize(v) for v in encoded_var] if not isinstance(encoded_var, dict): raise ValueError(f"The encoded_var should be dict and is {type(encoded_var)}") var = encoded_var[Encoding.VAR] type_ = encoded_var[Encoding.TYPE] if type_ == DAT.DICT: return {k: cls.deserialize(v) for k, v in var.items()} elif type_ == DAT.ASSET_EVENT_ACCESSORS: return _decode_outlet_event_accessors(var) elif type_ == DAT.ASSET_UNIQUE_KEY: return AssetUniqueKey(name=var["name"], uri=var["uri"]) elif type_ == DAT.ASSET_ALIAS_UNIQUE_KEY: return AssetAliasUniqueKey(name=var["name"]) elif type_ == DAT.DAG: return DagSerialization.deserialize_dag(var) elif type_ == DAT.OP: return OperatorSerialization.deserialize_operator(var) elif type_ == DAT.DATETIME: return from_timestamp(var) elif type_ == DAT.POD: # Attempt to import kubernetes for deserialization. Using attempt_import=True allows # lazy loading of kubernetes libraries only when actually needed for POD deserialization. if not _has_kubernetes(attempt_import=True): raise RuntimeError( "Cannot deserialize POD objects without kubernetes libraries. " "Please install the cncf.kubernetes provider." ) pod = PodGenerator.deserialize_model_dict(var) return pod elif type_ == DAT.TIMEDELTA: return datetime.timedelta(seconds=var) elif type_ == DAT.TIMEZONE: return parse_timezone(var) elif type_ == DAT.RELATIVEDELTA: return decode_relativedelta(var) elif type_ == DAT.AIRFLOW_EXC_SER or type_ == DAT.BASE_EXC_SER: deser = cls.deserialize(var) exc_cls_name = deser["exc_cls_name"] args = deser["args"] kwargs = deser["kwargs"] del deser if type_ == DAT.AIRFLOW_EXC_SER: exc_cls = import_string(exc_cls_name) else: exc_cls = import_string(f"builtins.{exc_cls_name}") return exc_cls(*args, **kwargs) elif type_ == DAT.BASE_TRIGGER: tr_cls_name, kwargs = cls.deserialize(var) tr_cls = import_string(tr_cls_name) return tr_cls(**kwargs) elif type_ == DAT.SET: return {cls.deserialize(v) for v in var} elif type_ == DAT.TUPLE: return tuple(cls.deserialize(v) for v in var) elif type_ == DAT.PARAM: return cls._deserialize_param(var) elif type_ == DAT.XCOM_REF: return _XComRef(var) # Delay deserializing XComArg objects until we have the entire DAG. elif type_ in (DAT.ASSET, DAT.ASSET_ALIAS, DAT.ASSET_ALL, DAT.ASSET_ANY, DAT.ASSET_REF): return decode_asset_like(encoded_var) elif type_ == DAT.CONNECTION: return Connection(**var) elif type_ == DAT.TASK_CALLBACK_REQUEST: return TaskCallbackRequest.from_json(var) elif type_ == DAT.DAG_CALLBACK_REQUEST: return DagCallbackRequest.from_json(var) elif type_ == DAT.TASK_INSTANCE_KEY: return TaskInstanceKey(**var) elif type_ == DAT.MAPPED_ARGUMENT: expand_input = create_expand_input(var["input"]["type"], var["input"]["value"]) return SchedulerMappedArgument(input=expand_input, key=var["key"]) elif type_ == DAT.ARG_NOT_SET: from airflow.serialization.definitions.notset import NOTSET return NOTSET elif type_ == DAT.DEADLINE_ALERT: return decode_deadline_alert(var) else: raise TypeError(f"Invalid type {type_!s} in deserialization.") === Airflow 3.3.0 BaseSerialization.deserialize === version 3.3.0 file /data/pruva/project-cache/03610ec6-e6fb-4086-9681-103dd9199da6/repo/airflow_3.3.0_cve_2026_33264_venv/lib/python3.14/site-packages/airflow/serialization/serialized_objects.py line 615 @classmethod def deserialize(cls, encoded_var: Any) -> Any: """ Deserialize an object; helper function of depth first search for deserialization. :meta private: """ if cls._is_primitive(encoded_var): return encoded_var elif isinstance(encoded_var, list): return [cls.deserialize(v) for v in encoded_var] if not isinstance(encoded_var, dict): raise ValueError(f"The encoded_var should be dict and is {type(encoded_var)}") var = encoded_var[Encoding.VAR] type_ = encoded_var[Encoding.TYPE] if type_ == DAT.DICT: return {k: cls.deserialize(v) for k, v in var.items()} elif type_ == DAT.ASSET_EVENT_ACCESSORS: return _decode_outlet_event_accessors(var) elif type_ == DAT.ASSET_UNIQUE_KEY: return AssetUniqueKey(name=var["name"], uri=var["uri"]) elif type_ == DAT.ASSET_ALIAS_UNIQUE_KEY: return AssetAliasUniqueKey(name=var["name"]) elif type_ == DAT.DAG: return DagSerialization.deserialize_dag(var) elif type_ == DAT.OP: return OperatorSerialization.deserialize_operator(var) elif type_ == DAT.DATETIME: return from_timestamp(var) elif type_ == DAT.POD: # Attempt to import kubernetes for deserialization. Using attempt_import=True allows # lazy loading of kubernetes libraries only when actually needed for POD deserialization. if not _has_kubernetes(attempt_import=True): raise RuntimeError( "Cannot deserialize POD objects without kubernetes libraries. " "Please install the `kubernetes` package." ) return deserialize_pod_dict(var) elif type_ == DAT.TIMEDELTA: return datetime.timedelta(seconds=var) elif type_ == DAT.TIMEZONE: return parse_timezone(var) elif type_ == DAT.RELATIVEDELTA: return decode_relativedelta(var) elif type_ == DAT.AIRFLOW_EXC_SER or type_ == DAT.BASE_EXC_SER: deser = cls.deserialize(var) exc_cls_name = deser["exc_cls_name"] args = deser["args"] kwargs = deser["kwargs"] del deser if type_ == DAT.AIRFLOW_EXC_SER: exc_cls = import_string(exc_cls_name) else: exc_cls = import_string(f"builtins.{exc_cls_name}") return exc_cls(*args, **kwargs) elif type_ == DAT.SET: return {cls.deserialize(v) for v in var} elif type_ == DAT.TUPLE: return tuple(cls.deserialize(v) for v in var) elif type_ == DAT.PARAM: return cls._deserialize_param(var) elif type_ == DAT.XCOM_REF: return _XComRef(var) # Delay deserializing XComArg objects until we have the entire DAG. elif type_ in (DAT.ASSET, DAT.ASSET_ALIAS, DAT.ASSET_ALL, DAT.ASSET_ANY, DAT.ASSET_REF): return decode_asset_like(encoded_var) elif type_ == DAT.CONNECTION: return Connection(**var) elif type_ == DAT.TASK_CALLBACK_REQUEST: return TaskCallbackRequest.from_json(var) elif type_ == DAT.DAG_CALLBACK_REQUEST: return DagCallbackRequest.from_json(var) elif type_ == DAT.TASK_INSTANCE_KEY: return TaskInstanceKey(**var) elif type_ == DAT.MAPPED_ARGUMENT: expand_input = create_expand_input(var["input"]["type"], var["input"]["value"]) return SchedulerMappedArgument(input=expand_input, key=var["key"]) elif type_ == DAT.ARG_NOT_SET: from airflow.serialization.definitions.notset import NOTSET return NOTSET elif type_ == DAT.DEADLINE_ALERT: return decode_deadline_alert(var) else: raise TypeError(f"Invalid type {type_!s} in deserialization.")