TaskMiddleware

Base Task Middleware.

TaskMiddleware

Bases: BaseMiddleware

TaskMiddleware - An abstract class that is the foundation for task Middleware classes for TaskExecutor.

Example

from qtasks.middlewares import TaskMiddleware
from qtasks.executors.base import BaseTaskExecutor

class MyTaskMiddleware(TaskMiddleware):
    def __init__(self, task_executor: BaseTaskExecutor):
        super().__init__(name="MyTaskMiddleware")
        self.task_executor = task_executor

Source code in src/qtasks/middlewares/task.py

class TaskMiddleware(BaseMiddleware):
    """
    `TaskMiddleware` - An abstract class that is the foundation for task Middleware classes for `TaskExecutor`.

    ## Example

    ```python
    from qtasks.middlewares import TaskMiddleware
    from qtasks.executors.base import BaseTaskExecutor

    class MyTaskMiddleware(TaskMiddleware):
        def __init__(self, task_executor: BaseTaskExecutor):
            super().__init__(name="MyTaskMiddleware")
            self.task_executor = task_executor
    ```
    """

    def __init__(self, task_executor: BaseTaskExecutor):
        """
        Initializing the task middleware.

        Args:
            task_executor (BaseTaskExecutor): A task executor instance.
        """
        super().__init__(name="TaskMiddleware")
        self.task_executor = task_executor

    def __call__(self, *args, **kwargs) -> Any:
        """
        Processing the task middleware call.

        Returns:
            Any: Processing result.
        """
        pass

__call__(*args, **kwargs)

Processing the task middleware call.

Returns:

Name	Type	Description
Any	Any	Processing result.

Source code in src/qtasks/middlewares/task.py

def __call__(self, *args, **kwargs) -> Any:
    """
    Processing the task middleware call.

    Returns:
        Any: Processing result.
    """
    pass

__init__(task_executor)

Initializing the task middleware.

Parameters:

Name	Type	Description	Default
task_executor	BaseTaskExecutor	A task executor instance.	required

Source code in src/qtasks/middlewares/task.py

def __init__(self, task_executor: BaseTaskExecutor):
    """
    Initializing the task middleware.

    Args:
        task_executor (BaseTaskExecutor): A task executor instance.
    """
    super().__init__(name="TaskMiddleware")
    self.task_executor = task_executor