Simplify asynchronous/long running callbacks

[FlorianJacta]

Description

Use Case:

Developers must use asynchronous callbacks to keep the client responsive (if callbacks take a long time to execute). This is an important concept for advanced use cases.

Current Challenges:

Currently, developers must rely on invoke_callback and invoke_long_callback. These functions are complex and require some knowledge to use:

invoke_callback requires:

• gui
• state_id (which requires understanding and using get_state_id())
• callback
• Additional arguments...

invoke_long_callback requires:

• state
• user_function
• user_function_args
• user_status_function
• user_status_function_args
• period (to specify how often the status function updates)

By itself, this API is not the simplest. This is somehow related to: Avaiga/taipy-doc#888

Although invoke_long_callback includes a period property for periodic updates, this does not provide real-time feedback on the progress of the heavy function. To achieve this, developers must implement additional logic, frequently invoking invoke_callback to update the state manually, as shown in the following example:

import time

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


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


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


def heavy_function(gui, state_id):
    value = invoke_callback(gui, state_id, user_status, ["Searching documents"])
    print(value)
    time.sleep(5)
    value = invoke_callback(gui, state_id, user_status, ["Responding to user"])
    print(value)
    time.sleep(5)
    value = invoke_callback(gui, state_id, user_status, ["Fact Checking"])
    time.sleep(5)
    print(value)
    return "Done"


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 = ""
    value = ""
    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()

This approach results in verbose, repetitive, and complex code.

Proposed Improvement:

The goal is to simplify the process for both:

1. Basic use cases — making a function asynchronous without blocking the client.
2. Advanced use cases — providing real-time feedback during long-running tasks.

The ideal solution would follow standard Taipy practices and avoid introducing overly complex concepts.

Initial Proposition:

Introduce an AsyncState class that mimics the behavior of a standard State but is designed for asynchronous operations. This simplifies the process for developers while maintaining flexibility.

Implementation:

AsyncState and Helper Functions:

from typing import Any
from taipy.gui import Gui, State, get_state_id, invoke_callback, invoke_long_callback


def run_async_function(state: State, heavy_function: Any):
    gui = state.get_gui()
    state_id = get_state_id(state)
    async_state = AsyncState(gui, state_id)
    invoke_long_callback(
        state=state,
        user_function=heavy_function,
        user_function_args=[async_state],
    )


def set_var_in_state(state: State, var_name: str, value: Any):
    setattr(state, var_name, value)


def get_var_from_state(state: State, var_name: str):
    return getattr(state, var_name)


class AsyncState:
    def __init__(self, gui: Gui, state_id: str) -> None:
        object.__setattr__(self, "async_gui", gui)
        object.__setattr__(self, "state_id", state_id)

    def __setattr__(self, var_name: str, var_value: Any) -> None:
        if var_name in {"async_gui", "state_id"}:
            object.__setattr__(self, var_name, var_value)
        else:
            invoke_callback(
                self.async_gui, self.state_id, set_var_in_state, [var_name, var_value]
            )

    def __getattr__(self, var_name: str) -> Any:
        if var_name in {"async_gui", "state_id"}:
            return object.__getattribute__(self, var_name)

        return invoke_callback(
            self.async_gui, self.state_id, get_var_from_state, [var_name]
        )

Developer Code:

from async_state import AsyncState, run_async_function
from taipy.gui import Gui
import taipy.gui.builder as tgb
import time


def heavy_function(state: AsyncState):
    state.logs = "Starting...\n"
    state.logs += "Searching documents\n"
    time.sleep(5)
    state.logs += "Responding to user\n"
    time.sleep(5)
    state.logs += "Fact Checking\n"
    time.sleep(5)
    state.result = "Done!"


def respond(state):
    run_async_function(state, heavy_function)


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()

Benefits of the Proposed Solution:

1. Simplified Code: Reduces the need for repetitive and verbose callback management.
2. Real-Time Feedback: Allows developers to easily provide live progress updates during execution.
3. Consistency: Follows Taipy's existing practices, ensuring a low learning curve for users.
4. Flexibility: Supports both simple and advanced use cases without introducing additional complexity.

Acceptance Criteria

• Any new code is covered by a unit tested.
• Check code coverage is at least 90%.

Code of Conduct

• I have checked the existing issues.
• I am willing to work on this issue (optional)