value: Any,
        *,
        key: str | None = None,
        task_id: str | None = None,
        dag_id: str | None = None,
        run_id: str | None = None,
        map_index: int | None = None,
    ) -> Any:
        """Serialize XCom value to str or pickled object."""
        if conf.getboolean("core", "enable_xcom_pickling"):
            return pickle.dumps(value)
        try:
            return json.dumps(value, cls=XComEncoder).encode("UTF-8")
        except (ValueError, TypeError) as ex:
            log.error(
                "%s."
                " If you are using pickle instead of JSON for XCom,"
                " then you need to enable pickle support for XCom"
                " in your airflow config or make sure to decorate your"
                " object with attr.",
                ex,
            )
            raise
    @staticmethod
    def _deserialize_value(result: XCom, orm: bool) -> Any: