def custom_schedule(
    is_cron: str, cron_schedule: str, cron_timezone: str
) -> Callable[[FlowSpecMeta], FlowSpecMeta]:
    """Decorator to add cron schedule to a flow."""

    def _custom_schedule(
        flow_spec_meta: FlowSpecMeta,
    ) -> FlowSpecMeta:
        """Decorator to add cron schedule to a flow."""
        if is_cron == "false":
            return flow_spec_meta

        return schedule(cron=cron_schedule, timezone=cron_timezone)(flow_spec_meta)

    return _custom_schedule

@custom_schedule(
    is_cron=config_expr("config.is_cron"),
    cron_schedule=config_expr("config.schedule"),
    cron_timezone=config_expr("config.timezone"),
)

def custom_engine_config(
        engine: str, local_mode: bool
    ) -> Callable[[FlowSpecMeta], FlowSpecMeta]:
        """Decorator to add engine config to a flow."""

        def _custom_engine_config(
            flow_spec_meta: FlowSpecMeta,
        ) -> FlowSpecMeta:
            """Decorator to add engine config to a flow."""
            if engine == "ray":
                if local_mode:
                    return kubernetes(
                        image=flow_constants.IMAGE,
                        cpu=4,
                        memory=24000,
                        qos="Guaranteed",
                    )(flow_spec_meta)

                return raystep(
                    address=flow_constants.RAY_CLUSTER_NAME,
                    timeout_seconds=flow_constants.RAY_TIMEOUT_SECONDS,
                    env_vars=flow_constants.RAY_ENV_VARS,
                )(flow_spec_meta)

            raise ValueError(f"Invalid engine {engine}. Supported engines are: ray.")

ValueError: Invalid engine <metaflow.user_configs.config_parameters.DelayEvaluator object at 0x1047b92e0>. Supported engines are: ray.