Task Management

def task(callee: Callable[..., Any]) -> Callable[..., Optional[Union[Thread, Process]]]:
    """
    Decorator that converts a function into an asynchronous task.
    
    This is the main decorator of the library. It wraps any function
    to make it run asynchronously in the background using the current
    pool's configuration (threads or processes).
    
    Args:
        callee: The function to be made asynchronous
    
    Returns:
        Decorated function that returns Thread/Process object or None
    """

import multitasking
import time

@multitasking.task
def process_item(item_id):
    # Simulate processing time
    time.sleep(1)
    return f"Processed item {item_id}"

# Start multiple tasks
for i in range(5):
    task_obj = process_item(i)
    print(f"Started task: {task_obj}")

# Wait for all to complete
multitasking.wait_for_tasks()

def get_list_of_tasks() -> List[Union[Thread, Process]]:
    """
    Retrieve all tasks ever created by this library.
    
    This includes both currently running tasks and completed ones.
    Useful for debugging and monitoring task history.
    
    Returns:
        List of all Thread/Process objects created by @task decorator
    
    Note:
        Completed tasks remain in this list until program termination.
    """

import multitasking

@multitasking.task
def worker(task_id):
    time.sleep(1)
    print(f"Task {task_id} completed")

# Start some tasks
for i in range(3):
    worker(i)

# Check all tasks (running and completed)
all_tasks = multitasking.get_list_of_tasks()
print(f"Total tasks created: {len(all_tasks)}")

multitasking.wait_for_tasks()

# All tasks still listed even after completion
final_tasks = multitasking.get_list_of_tasks()
print(f"Tasks after completion: {len(final_tasks)}")

def get_active_tasks() -> List[Union[Thread, Process]]:
    """
    Retrieve only the currently running tasks.
    
    Filters the complete task list to show only tasks that are still
    executing. This is more useful than get_list_of_tasks() for
    monitoring current system load.
    
    Returns:
        List of Thread/Process objects that are still running
    """

import multitasking
import time

@multitasking.task
def long_task(duration):
    time.sleep(duration)
    print(f"Task completed after {duration}s")

# Start tasks with different durations
long_task(3)
long_task(2)
long_task(1)

# Monitor progress
while multitasking.get_active_tasks():
    active_count = len(multitasking.get_active_tasks())
    total_count = len(multitasking.get_list_of_tasks())
    completed = total_count - active_count
    print(f"Progress: {completed}/{total_count} completed")
    time.sleep(0.5)

print("All tasks finished!")