15 task manager

TaskManagerService - Task Orchestration API

Central API for task management, scheduling, và monitoring

Overview

TaskManagerService cung cấp unified API cho:

• Task execution
• Scheduling (date, interval, cron)
• Task chains
• Status monitoring
• Persistence

API Reference

Access

from core import QtAppContext

ctx = QtAppContext.globalInstance()
taskManager = ctx.taskManager

Add Task

taskManager.addTask(task, scheduleInfo=None)

Add Chain

chain = taskManager.addChainTask(
    name='Chain Name',
    tasks=[task1, task2],
    retryBehaviorMap={...}
)

Cancel Task

taskManager.cancelTask(uuid)

Bulk Actions

# Stop all tasks targeted by tag
taskManager.stopTasksByTag('Network')
# Include chained children (danger zone)
taskManager.stopTasksByTag('Network', includeChainedChildren=True)

# Pause all tasks targeted by tag
taskManager.pauseTasksByTag('HeavyComputation')

Get Status

status = taskManager.getTaskStatus(uuid)
allTasks = taskManager.getAllTasks()
failedTasks = taskManager.getFailedTasks()

Configuration

taskManager.setMaxConcurrentTasks(5)
queueStatus = taskManager.getQueueStatus()

Scheduling

Date-based

from datetime import datetime, timedelta

taskManager.addTask(task, scheduleInfo={
    'trigger': 'date',
    'runDate': datetime.now() + timedelta(hours=1)
})

Interval

taskManager.addTask(task, scheduleInfo={
    'trigger': 'interval',
    'intervalSeconds': 60  # Every minute
})

Cron

taskManager.addTask(task, scheduleInfo={
    'trigger': 'cron',
    'hour': 9,
    'minute': 0,
    'day_of_week': 'mon-fri'
})

Signals

taskManager.taskAdded.connect(lambda uuid: print(f'Added: {uuid}'))
taskManager.taskRemoved.connect(lambda uuid: print(f'Removed: {uuid}'))
taskManager.statusChanged.connect(lambda uuid, status: print(f'{uuid}: {status}'))
taskManager.progressUpdated.connect(lambda uuid, progress: print(f'{uuid}: {progress}%'))

Usage Examples

Basic Task Execution

from core import QtAppContext
from core.taskSystem import AbstractTask

class MyTask(AbstractTask):
    def handle(self):
        # Task logic...
        pass
    
    def _performCancellationCleanup(self):
        pass

ctx = QtAppContext.globalInstance()
taskManager = ctx.taskManager

task = MyTask(name='My Task')
taskManager.addTask(task)

Scheduled Task

from datetime import datetime, timedelta

# Run in 1 hour
task = MyTask(name='Scheduled Task')
taskManager.addTask(task, scheduleInfo={
    'trigger': 'date',
    'runDate': datetime.now() + timedelta(hours=1)
})

# Run every 5 minutes
task = MyTask(name='Periodic Task')
taskManager.addTask(task, scheduleInfo={
    'trigger': 'interval',
    'intervalSeconds': 300
})

Task Chain

chain = taskManager.addChainTask(
    name='Data Pipeline',
    tasks=[
        FetchTask(name='Fetch'),
        ProcessTask(name='Process'),
        SaveTask(name='Save')
    ],
    retryBehaviorMap={
        'Fetch': ChainRetryBehavior.RETRY_TASK,
        'Process': ChainRetryBehavior.SKIP_TASK,
        'Save': ChainRetryBehavior.FAIL_CHAIN
    }
)

Monitoring

# Monitor all tasks
def onStatusChanged(uuid, status):
    print(f'Task {uuid}: {status}')

def onProgress(uuid, progress):
    print(f'Task {uuid}: {progress}%')

taskManager.statusChanged.connect(onStatusChanged)
taskManager.progressUpdated.connect(onProgress)

# Get queue status
status = taskManager.getQueueStatus()
print(f'Active: {status["active"]}, Pending: {status["pending"]}')

Cancel Task

# Add task
task = MyTask(name='Cancellable Task')
taskManager.addTask(task)

# Cancel later
taskManager.cancelTask(task.uuid)

Best Practices

✅ DO

# Check if tasks enabled
ctx = QtAppContext.globalInstance()
if ctx.isFeatureEnabled('tasks'):
    taskManager = ctx.taskManager

# Monitor task status
taskManager.statusChanged.connect(onStatusChanged)

# Set appropriate max concurrent
taskManager.setMaxConcurrentTasks(3)

# Use descriptive names
task = MyTask(name='Download User Data')

❌ DON'T

# Don't assume tasks enabled
taskManager = ctx.taskManager  # May be None!

# Don't create too many concurrent tasks
taskManager.setMaxConcurrentTasks(100)  # Too many!

# Don't forget to handle failures
# Monitor failedTasks and handle appropriately

Related Documentation

• AbstractTask - Task implementation
• TaskChain - Task chaining
• Task System Overview - Architecture