Function run_process_with_fallbacks

reflex/utils/processes.py:486–538

Run subprocess and retry using fallback command if initial command fails. Args: args: A string, or a sequence of program arguments. show_status_message: The status message to be displayed in the console. fallbacks: The fallback command to run if the initial command fails

(
    args: list[str],
    *,
    show_status_message: str,
    fallbacks: str | Sequence[str] | Sequence[Sequence[str]] | None = None,
    analytics_enabled: bool = False,
    prior_logs: tuple[tuple[str, ...], ...] = (),
    **kwargs,
)

def run_process_with_fallbacks(
    args: list[str],
    *,
    show_status_message: str,
    fallbacks: str | Sequence[str] | Sequence[Sequence[str]] | None = None,
    analytics_enabled: bool = False,
    prior_logs: tuple[tuple[str, ...], ...] = (),
    **kwargs,
):
    """Run subprocess and retry using fallback command if initial command fails.

    Args:
        args: A string, or a sequence of program arguments.
        show_status_message: The status message to be displayed in the console.
        fallbacks: The fallback command to run if the initial command fails.
        analytics_enabled: Whether analytics are enabled for this command.
        prior_logs: The logs of the prior processes that have been run.
        **kwargs: Kwargs to pass to new_process function.
    """
    process = new_process(get_command_with_loglevel(args), **kwargs)
    if not fallbacks:
        # No fallback given, or this _is_ the fallback command.
        show_status(
            show_status_message,
            process,
            analytics_enabled=analytics_enabled,
            prior_logs=prior_logs,
        )
    else:
        # Suppress errors for initial command, because we will try to fallback
        logs = show_status(show_status_message, process, suppress_errors=True)

        current_fallback = fallbacks[0] if not isinstance(fallbacks, str) else fallbacks
        next_fallbacks = fallbacks[1:] if not isinstance(fallbacks, str) else None

        if process.returncode != 0:
            # retry with fallback command.
            fallback_with_args = (
                [current_fallback, *args[1:]]
                if isinstance(current_fallback, str)
                else [*current_fallback, *args[1:]]
            )
            console.warn(
                f"There was an error running command: {args}. Falling back to: {fallback_with_args}."
            )
            run_process_with_fallbacks(
                fallback_with_args,
                show_status_message=show_status_message,
                fallbacks=next_fallbacks,
                analytics_enabled=analytics_enabled,
                prior_logs=(*prior_logs, tuple(logs)),
                **kwargs,
            )


def execute_command_and_return_output(command: str) -> str | None: