Skip to content

Long Running Callbacks

Tasks (server-side functions) in web applications can take a lot of time, which can lead to problems with communication between the server (Python application) and the client (web browser).

See example code

Taipy offers a valuable feature known as "long-running callbacks" to tackle this problem. These callbacks enable the server to handle resource-intensive processing in the background, ensuring a responsive user interface.

Long Running Callbacks

This article discusses the concept of long-running callbacks in Taipy, provides usage examples, and illustrates how they enhance the overall user experience.

Running Functions in Background

Imagine a situation where a callback starts a duty that requires a lot of resources and time to finish. To make this work, we can use a straightforward approach:

    from taipy.gui import State, invoke_long_callback, notify

    def heavy_function(...):
        # Do something that takes time...
        ...

    def on_action(state):
        notify(state, "info", "Heavy function started")
        invoke_long_callback(state, heavy_function, [...heavy_function arguments...])

In the previous example, the Taipy function invoke_long_callback() manages the resource-intensive function. It sets up a separate background thread to run heavy_function(), allowing the rest of the application to continue running. The on_action() function gets activated by a user action, like clicking a button.

Monitoring Function Status

Moreover, you can send notifications to the user's browser or update visual elements depending on the status of the ongoing process. Taipy offers a way to receive notifications when the function completes, as shown below:

    def heavy_function_status(state, status):
        if status:
            notify(state, "success", "The heavy function has finished!")
        else:
            notify(state, "error", "The heavy function has failed")

    def on_action(state, id, action):
        invoke_long_callback(state, heavy_function, [...heavy_function arguments...],
                             heavy_function_status)

In this example, we introduce the heavy_function_status() function, which the invoke_long_callback()^ function invokes. When the callback is finished, this function is called.

This allows you to provide the necessary notifications or make updates to the user interface depending on whether the processing was successful or not.

Handling Function Result

To update the State according to the returned value from heavy_function(), you can modify heavy_function_status() as follows:

1
2
3
4
5
6
7
    def heavy_function_status(state, status, result):
        if status:
            notify(state, "success", "The heavy function has finished!")
            # Actualize the State with the function result
            state.result = result
        else:
            notify(state, "error", "The heavy function has failed")

We added a parameter called result, which represents the return value of heavy_function(). When heavy_function() completes successfully (status is True), we update the State with the result by assigning it to a state variable (cf. line 5). This allows you to access the result in other parts of your application or display it to the user as needed.

Make sure that the heavy_function() returns a value. For example:

    def heavy_function(...):
        ...
        return result

When you update the State with the result of heavy_function(), you ensure that the user interface shows the result of the resource-intensive function. This creates a smooth and seamless user experience.

Tracking Function Progress

Trigger update from heavy_function

In some cases, it is beneficial to trigger updates from within the heavy_function itself. This can be done using the invoke_callback function to send updates to the client during the execution of the long-running task.

Here's an example:

import time

import taipy.gui.builder as tgb
from taipy.gui import Gui, get_state_id, invoke_callback, invoke_long_callback


def status_fct(state, status, result):
    state.logs = ""
    state.result = result

def user_status(state, info):
    state.logs = state.logs + "\n" + info

def heavy_function(gui, state_id):
    invoke_callback(gui, state_id, user_status, ["Searching documents"])
    time.sleep(5)
    invoke_callback(gui, state_id, user_status, ["Responding to user"])
    time.sleep(5)
    invoke_callback(gui, state_id, user_status, ["Fact Checking"])
    return "Here is the answer"

def respond(state):
    invoke_long_callback(state=state,
                         user_function=heavy_function, user_function_args=[gui, get_state_id(state)],
                         user_status_function=status_fct, user_status_function_args=[])

if __name__ == "__main__":
    logs = ""
    result = "No response yet"

    with tgb.Page() as main_page:
        tgb.button("Respond", on_action=respond)
        with tgb.part("card"):
            tgb.text("{logs}", mode="pre")

        tgb.text("# Result", mode="md")
        tgb.text("{result}")


    gui = Gui(main_page)
    gui.run()

In this example, the heavy_function uses the invoke_callback function to send updates to the client at different stages of the task. The user_status function appends these updates to the logs state variable, which is then displayed in the user interface.

  1. heavy_function:
  2. It calls invoke_callback at different stages to send progress updates to the user_status function.
  3. After completing the task, it returns the result.

  4. user_status:

  5. It updates the logs state variable with the progress information.

  6. status_fct:

  7. It updates the result state variable with the final result of the heavy_function.

  8. respond:

  9. It initiates the long-running task by calling invoke_long_callback with the heavy_function and associated status function.

By using this approach, you can provide real-time updates to the user interface directly from within the heavy_function, enhancing the user experience by keeping them informed about the progress of the long-running task.

Regular Updates with Time Intervals

Occasionally, it's useful to give regular updates on the progress of a long-running task. Taipy's invoke_long_callback() provides a convenient method to accomplish this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
    def heavy_function_status(state, status):
        if isinstance(status, bool):
            if status:
                notify(state, "success", "The heavy function has finished!")
            else:
                notify(state, "error", "The heavy function has failed somehow.")
        else:
            notify(state, "info", "The heavy function is still running...")

    def on_action(state):
        invoke_long_callback(state, heavy_function, [...heavy_function arguments...],
                             heavy_function_status, [...heavy_function_status arguments...],
                             5000)

In the code above, in line 13, when you include a period parameter, the heavy_function_status() function will be regularly activated at the set interval, such as every 5 seconds. This allows your user interface to show live updates, informing the end user about ongoing work.

Conclusion and code

Taipy's long-running callbacks make handling time-consuming tasks in web applications much easier. By running demanding functions in the background, Taipy ensures that the user interface responds quickly and avoids potential communication timeouts. With the option to keep an eye on the function's progress and offer updates, developers can create a smooth user experience, even when dealing with hefty operations.

Approximating Pi

from taipy.gui import Gui, Markdown, State, invoke_long_callback, notify


def pi_approx(num_iterations: int):
    """
    Approximate Pi using the Leibniz formula.

    Args:
        num_iterations: Number of iterations to compute the approximation.

    Returns:
        A list of approximations of Pi, made at each iteration.
    """
    k, s = 3.0, 1.0
    pi_list = []
    for i in range(num_iterations):
        s = s - ((1 / k) * (-1) ** i)
        k += 2
        if (i + 1) % (int(num_iterations / 100) + 1) == 0:
            pi_list += [4 * s]

    return pi_list


def heavy_status(state: State, status, pi_list: list):
    """
    Periodically update the status of the long callback.

    Args:
        state: The state of the application.
        status: The status of the long callback.
        pi_list: The list of approximations of Pi.
    """
    state.logs = f"Approximating Pi... ({status}s)"
    if isinstance(status, bool):
        if status:
            state.logs = f"Finished! Approximation: {pi_list[-1]}"
            notify(state, "success", "Finished")
            state.pi_list = pi_list
        else:
            notify(state, "error", "An error was raised")
    else:
        state.status += 1


def on_action(state: State):
    """
    When the button is clicked, start the long callback.

    Args:
        state: The state of the application.
    """
    invoke_long_callback(
        state, pi_approx, [int(state.num_iterations)], heavy_status, [], 1000
    )

if __name__ == "__main__":
    status = 0
    num_iterations = 20_000_000
    pi_list = []
    logs = "Not running"

    page = Markdown("""
# Approximating **Pi**{: .color-primary} using the Leibniz formula
<|{num_iterations}|number|label=Number of iterations|>
<|Approximate Pi|button|on_action=on_action|>
## Evolution of approximation
<|{pi_list}|chart|layout={layout}|>
<br/>
<|card|
## Logs
### <|{logs}|text|raw|>
|>
    """)

    layout = {
        "xaxis": {"title": "Iteration (Percentage of Total Iterations)"},
        "yaxis": {"title": "Pi Approximation"},
    }

    Gui(page).run()