Skip to content

Tasker Adapter

HydratorTasker

Manages tasks generated by the Zenaura diffing algorithm for updating both the real DOM and the virtual DOM.

This class provides methods for:

  • Creating and retrieving task queues for specific component IDs.
  • Enqueuing tasks for components.
  • Dequeuing tasks for components.

Key Features:

  • Efficient task management for real and virtual DOM updates.
  • Thread-safe task queue operations.
  • Automatic cleanup of empty task queues.

Usage:

  1. Use hyd_tsk_get_or_create_task_queue to get or create a task queue for a specific component ID.
  2. Use hyd_tsk_enqueue_task to enqueue a task for a component.
  3. Use hyd_tsk_dequeue_task to dequeue a task for a component.

Example:

tasker = HydratorTasker()

# Get or create a task queue for component ID "my-component"
task_queue = tasker.hyd_tsk_get_or_create_task_queue("my-component")

# Enqueue a task for the component
task_queue.put_nowait(my_task)

# Dequeue a task for the component
task = task_queue.get_nowait()

# Execute the task
task()
Source code in zenaura/client/hydrator/tasker.py 
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class HydratorTasker:
    """
    Manages tasks generated by the Zenaura diffing algorithm for updating both the real DOM and the virtual DOM.

    This class provides methods for:

    - Creating and retrieving task queues for specific component IDs.
    - Enqueuing tasks for components.
    - Dequeuing tasks for components.

    **Key Features:**

    - Efficient task management for real and virtual DOM updates.
    - Thread-safe task queue operations.
    - Automatic cleanup of empty task queues.

    **Usage:**

    1. Use `hyd_tsk_get_or_create_task_queue` to get or create a task queue for a specific component ID.
    2. Use `hyd_tsk_enqueue_task` to enqueue a task for a component.
    3. Use `hyd_tsk_dequeue_task` to dequeue a task for a component.

    **Example:**

    ```python
    tasker = HydratorTasker()

    # Get or create a task queue for component ID "my-component"
    task_queue = tasker.hyd_tsk_get_or_create_task_queue("my-component")

    # Enqueue a task for the component
    task_queue.put_nowait(my_task)

    # Dequeue a task for the component
    task = task_queue.get_nowait()

    # Execute the task
    task()
    ```
    """

    queue_lookup = defaultdict(lambda: None)

    def __init__(self):
        """
        Initializes the HydratorTasker instance.
        """
        pass

    def hyd_tsk_get_or_create_task_queue(self, component_id) -> asyncio.Queue:
        """
        Gets or creates a task queue for the specified component ID.

        Args:
            component_id (str): The ID of the component.

        Returns:
            asyncio.Queue: The task queue for the component.
        """
        try:
            if component_id not in self.queue_lookup:
                self.queue_lookup[component_id] = asyncio.Queue()
                return self.queue_lookup[component_id]
            else:
                return self.queue_lookup[component_id]
        except KeyError:
            return False

    def hyd_tsk_enqueue_task(self, component_id, task):
        """
        Enqueues a task for the specified component ID.

        Args:
            component_id (str): The ID of the component.
            task (callable): The task to be enqueued.

        Returns:
            bool: True if the task was successfully enqueued, False otherwise.
        """
        comp_queue = self.queue_lookup[component_id]
        if not comp_queue:
            return False 
        try:
            comp_queue.put_nowait(task)
            return True
        except asyncio.QueueFull:
            return False

    async def hyd_tsk_do_nothing(self):
        """
        A placeholder function used when a task queue is empty.
        """
        return asyncio.Queue()

    def hyd_tsk_dequeue_task(self, component_id):
        """
        Dequeues a task for the specified component ID.

        Args:
            component_id (str): The ID of the component.

        Returns:
            callable: The dequeued task, or `hyd_tsk_do_nothing` if the queue is empty.
        """

        comp_queue = self.queue_lookup[component_id]
        if not comp_queue:
            return self.hyd_tsk_do_nothing
        try:
            task = comp_queue.get_nowait()
            return task
        except asyncio.QueueEmpty:
            # Clean up and return the placeholder function
            self.queue_lookup[component_id] = asyncio.Queue()
            return self.hyd_tsk_do_nothing

__init__()

Initializes the HydratorTasker instance.

Source code in zenaura/client/hydrator/tasker.py 
47
48
49
50
51
def __init__(self):
    """
    Initializes the HydratorTasker instance.
    """
    pass

hyd_tsk_dequeue_task(component_id)

Dequeues a task for the specified component ID.

Parameters:

Name Type Description Default
component_id str

The ID of the component.

 required

Returns:

Name Type Description
callable

The dequeued task, or hyd_tsk_do_nothing if the queue is empty.
Source code in zenaura/client/hydrator/tasker.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def hyd_tsk_dequeue_task(self, component_id):
    """
    Dequeues a task for the specified component ID.

    Args:
        component_id (str): The ID of the component.

    Returns:
        callable: The dequeued task, or `hyd_tsk_do_nothing` if the queue is empty.
    """

    comp_queue = self.queue_lookup[component_id]
    if not comp_queue:
        return self.hyd_tsk_do_nothing
    try:
        task = comp_queue.get_nowait()
        return task
    except asyncio.QueueEmpty:
        # Clean up and return the placeholder function
        self.queue_lookup[component_id] = asyncio.Queue()
        return self.hyd_tsk_do_nothing

hyd_tsk_do_nothing() async

A placeholder function used when a task queue is empty.

Source code in zenaura/client/hydrator/tasker.py 
92
93
94
95
96
async def hyd_tsk_do_nothing(self):
    """
    A placeholder function used when a task queue is empty.
    """
    return asyncio.Queue()

hyd_tsk_enqueue_task(component_id, task)

Enqueues a task for the specified component ID.

Parameters:

Name Type Description Default
component_id str

The ID of the component.

 required
task callable

The task to be enqueued.

 required

Returns:

Name Type Description
bool

True if the task was successfully enqueued, False otherwise.
Source code in zenaura/client/hydrator/tasker.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def hyd_tsk_enqueue_task(self, component_id, task):
    """
    Enqueues a task for the specified component ID.

    Args:
        component_id (str): The ID of the component.
        task (callable): The task to be enqueued.

    Returns:
        bool: True if the task was successfully enqueued, False otherwise.
    """
    comp_queue = self.queue_lookup[component_id]
    if not comp_queue:
        return False 
    try:
        comp_queue.put_nowait(task)
        return True
    except asyncio.QueueFull:
        return False

hyd_tsk_get_or_create_task_queue(component_id)

Gets or creates a task queue for the specified component ID.

Parameters:

Name Type Description Default
component_id str

The ID of the component.

 required

Returns:

Type Description
Queue

asyncio.Queue: The task queue for the component.
Source code in zenaura/client/hydrator/tasker.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def hyd_tsk_get_or_create_task_queue(self, component_id) -> asyncio.Queue:
    """
    Gets or creates a task queue for the specified component ID.

    Args:
        component_id (str): The ID of the component.

    Returns:
        asyncio.Queue: The task queue for the component.
    """
    try:
        if component_id not in self.queue_lookup:
            self.queue_lookup[component_id] = asyncio.Queue()
            return self.queue_lookup[component_id]
        else:
            return self.queue_lookup[component_id]
    except KeyError:
        return False