quickie.factories.thread_group

quickie.factories.thread_group(fn: Callable) → type[ThreadGroup]
quickie.factories.thread_group(*, name: str | None = None, extra_args: bool | None = False, bind: bool = False, condition: BaseCondition | None = None, before: Sequence[Any] | None = None, after: Sequence[Any] | None = None, cleanup: Sequence[Any] | None = None) → PartialReturnType[ThreadGroup]

Create a thread group task from a function.

The returned task will run in parallel. To add arguments to individual tasks in the group, you can return an instance of partial_task with the task and the arguments.

Note that the tasks run in separate threads, so they should be thread-safe. This means that they are also affected by the Global Interpreter Lock (GIL).

@thread_group
@arg("arg1")
def my_group(arg1):
    return [task1, partial_task(task2, arg1)]
Parameters:

  • fn – The function to create the thread group task from.

  • name – The name of the thread group task.

  • extra_args – If the thread group task accepts extra arguments.

  • bind – If true, the first parameter of the function will be the task class instance.

  • condition – The condition to check before running the thread group task.

  • before – The tasks to run before the thread group task.

  • after – The tasks to run after the thread group task.

  • cleanup – The tasks to run after the thread group task, even if it fails.

Returns:

The thread group task class, or, if fn is None, a partial function to be used as a decorator for a function.