Custom Task

Quantum methods for running custom tasks.

CustomTask

Quantum methods for running custom tasks.

Source code in src/scruby/mixins/custom_task.py

class CustomTask:
    """Quantum methods for running custom tasks."""

    @final
    async def run_custom_task(
        self,
        custom_task_fn: Callable,
        filter_fn: Callable = lambda _: True,
        **kwargs,
    ) -> Any:
        """For run an asynchronous custom task.

        Attention:
            - The search is based on the effect of a quantum loop.
            - The search effectiveness depends on the number of processor threads.

        Args:
            custom_task_fn (Callable): A function that execute the custom task.

        Returns:
            The result of a custom task.
        """
        return await custom_task_fn(
            search_task_fn=self._task_find,
            filter_fn=filter_fn,
            branch_numbers=range(self._max_number_branch),
            hash_reduce_left=self._hash_reduce_left,
            db_root=self._db_root,
            class_model=self._class_model,
            max_workers=self._max_workers,
            **kwargs,
        )

run_custom_task(custom_task_fn, filter_fn=lambda _: True, **kwargs) async

For run an asynchronous custom task.

Attention

• The search is based on the effect of a quantum loop.
• The search effectiveness depends on the number of processor threads.

Parameters:

Name	Type	Description	Default
custom_task_fn	Callable	A function that execute the custom task.	required

Returns:

Type	Description
Any	The result of a custom task.

Source code in src/scruby/mixins/custom_task.py

@final
async def run_custom_task(
    self,
    custom_task_fn: Callable,
    filter_fn: Callable = lambda _: True,
    **kwargs,
) -> Any:
    """For run an asynchronous custom task.

    Attention:
        - The search is based on the effect of a quantum loop.
        - The search effectiveness depends on the number of processor threads.

    Args:
        custom_task_fn (Callable): A function that execute the custom task.

    Returns:
        The result of a custom task.
    """
    return await custom_task_fn(
        search_task_fn=self._task_find,
        filter_fn=filter_fn,
        branch_numbers=range(self._max_number_branch),
        hash_reduce_left=self._hash_reduce_left,
        db_root=self._db_root,
        class_model=self._class_model,
        max_workers=self._max_workers,
        **kwargs,
    )