TaskManager - Сервис для запуска и конфигурирования пользовательских задач

class axipy.concurrent.TaskManager(progress_element_factory=<axipy.concurrent.TaskManager.ProgressElementFactory object>)

Сервис, содержащий вспомогательные функции для запуска и конфигурирования пользовательских задач.

generate_dialog_for_task(task, spec)

Возвращает диалог, следящий за выполнением задачи.

Параметры

  • task (AxipyTask) – Пользовательская задача.

  • spec (ProgressSpecification) – Описание задачи.

Тип результата

QDialog

Результат

Диалог, который будет отображать прогресс выполнения задачи.

Пример использования.
def user_heavy_func(ph: AxipyProgressHandler, arg1: str, arg2: str):
    print(arg1 + arg2)

# Создаём задачу из пользовательской функции и передаём необходимые
# аргументы
task = AxipyAnyCallableTask(user_heavy_func, "Длительная ", "задача")
spec = ProgressSpecification(description="Длительная задача")
dialog = task_manager.generate_dialog_for_task(task, spec)
# Ставим задачу в очередь на выполнение
task_manager.start_task(task)
# Показываем диалог с прогрессом пока не завершилась задача
dialog.exec_()
run_and_get(spec, func, *args, **kwargs)

Превращает пользовательскую функцию в задачу и запускает её с отображением прогресса выполнения.

Параметры

  • spec (ProgressSpecification) – Описание задачи.

  • func (Callable) – Пользовательская функция, которая будет выполняться. В нее передается список args и словарь kwargs.

  • args – Список аргументов, передаваемый в функцию при запуске.

  • kwargs – Словарь, передаваемый в функцию при запуске.

Тип результата

Any

Результат

Результат выполнения пользовательской функции или None при ошибке.

Пример использования.
def user_heavy_function(ph: AxipyProgressHandler, count: int):
    # Вначале задаём верхнюю планку изменения прогресса
    ph.set_max_progress(count)
    for i in range(0, count):
        if ph.is_canceled():
            break
        # Тут делаем длительные вычисления
        ph.add_progress(1)
    return ph.progress()

spec = ProgressSpecification(
    description="Длительная операция",
    flags=ProgressGuiFlags.CANCELABLE)
times = 6
real_times = task_manager.run_and_get(spec, user_heavy_function, times)
# выводим колличество раз которое отработал цикл внутри
print(real_times)
run_in_gui(func, *args, **kwargs)

Выполняет переданную задачу в потоке интерфейса.

Это может быть удобно когда в процессе выполнения длительной фоновой задачи нужно спросить о чем нибудь пользователя отобразив диалог. Так же создавать/взаимодействовать с некоторыми объектами можно только из потока интерфейса.

Тип результата

Any

start_task(task)

Добавляет переданную задачу в очередь на выполнение.

Параметры

task – Пользовательская задача.

Пример использования.
def user_function(message: str):
    print(message)
task = AxipyAnyCallableTask(user_function, "Hi, world!")
# Т.к. мы не управляем прогрессом, то можно отключить передачу
# обработчика в функцию
task.with_handler(False)
task_manager.start_task(task)