reflex_copy/reflex/utils/redir.py

"""Utilities to handle redirection to browser UI."""

import time
import uuid
import webbrowser

import httpx

from .. import constants
from . import console


def open_browser_and_wait(
    target_url: str, poll_url: str, interval: int = 1
) -> httpx.Response:
    """Open a browser window to target_url and request poll_url until it returns successfully.

    Args:
        target_url: The URL to open in the browser.
        poll_url: The URL to poll for success.
        interval: The interval in seconds to wait between polling.

    Returns:
        The response from the poll_url.
    """
    if not webbrowser.open(target_url):
        console.warn(
            f"Unable to automatically open the browser. Please navigate to {target_url} in your browser."
        )
    console.info("Complete the workflow in the browser to continue.")
    while response := httpx.get(poll_url, follow_redirects=True):
        if response.is_success:
            break
        time.sleep(interval)
    return response


def reflex_build_redirect() -> str:
    """Open the browser window to reflex.build and wait for the user to select a generation.

    Returns:
        The selected generation hash.
    """
    token = str(uuid.uuid4())
    target_url = constants.Templates.REFLEX_BUILD_URL.format(reflex_init_token=token)
    poll_url = constants.Templates.REFLEX_BUILD_POLL_URL.format(reflex_init_token=token)
    response = open_browser_and_wait(target_url, poll_url)
    return response.json()["generation_hash"]