ThreadSet

A container for sets of threads. It provides boilerplate-removing utilities for handling standard library threads. See Python's threading.Thread documentation for general information on Python threads.

__init__(*threads)

You can initialize a ThreadSet with any threads you already defined:

class MyCustomThreadSubclass(threading.Thread):
    ...

threads = ThreadSet(
    MyCustomThreadSubclass(...),
    threading.Thread(...),
)
threads.start_and_join()

You can create unions of ThreadSets:

readers = ThreadSet(...)
readers.start()

writers = ThreadSet(...)
writers.start()

(readers | writers).join()

You can create a ThreadSet in a function definition, using with_args(), repeat(), or target:

@ThreadSet.repeat(10)  # will create 10 threads
def workers():
    ...

workers.start_and_join()

With ThreadSet.target, function typing information is kept:

@ThreadSet.target
def reader(thread_id: int, color: str):
    ...

@ThreadSet.target
def writer(thread_id: int, color: str):
    ...

threads = ThreadSet(
    reader(0, "red"),  # your IDE will show the correct types here
    reader(1, "green"),
    writer(2, "blue")
)
threads.start_and_join()

with_args(*args) classmethod

@ThreadSet.with_args(ThreadSet.Args(1, color="red"), ThreadSet.Args(2, "blue"))
def spam(thread_id, color):
    ...

spam.start_and_join()

repeat(times) classmethod

@ThreadSet.repeat(5)
def workers():
    ...

workers.start_and_join()

target(target) classmethod

@ThreadSet.target
def spam(thread_id: int, color: str):
    ...

threads = ThreadSet(
    spam(1, color="red"),
    spam(2, "blue"),
)
threads.start_and_join()

start()

Start the threads in this ThreadSet.

Also see Thread.start().

join(timeout=None)

Join the threads in this ThreadSet.

Also see Thread.join().

Parameters:

Name	Type	Description	Default
timeout	float | None	The timeout for each individual join to complete.	None

start_and_join(join_timeout=None)

Start the threads in this ThreadSet, then join them.

is_alive()

Call Thread.is_alive() for each thread in this ThreadSet.

any_is_alive()

any(self.is_alive())

all_are_alive()

all(self.is_alive())

all_are_not_alive()

not self.any_is_alive()

__or__(other)

Returns a new ThreadSet containing all the threads in the operands, which must be ThreadSet instances.

threads = ThreadSet(...) | ThreadSet(...)

__ior__(other)

Adds the threads in the right operand into this ThreadSet. The right operand must be a ThreadSet.

threads = ThreadSet(...)
threads |= ThreadSet(...)

__len__()

Returns the number of threads contained in this ThreadSet.

len(ThreadSet(...))