ThreadSet

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.

Methods:

• __init__ –

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

• with_args –

  ```python

• repeat –

  ```python

• target –

  ```python

• start –

  Start the threads in this ThreadSet.

• join –

  Join the threads in this ThreadSet.

• start_and_join –

  Start the threads in this ThreadSet, then join them.

• is_alive –

  Call Thread.is_alive()

• any_is_alive –

  ```python

• all_are_alive –

  ```python

• all_are_not_alive –

  ```python

• __or__ –

  Returns a new ThreadSet containing all the threads in the operands,

• __ior__ –

  Adds the threads in the right operand into this ThreadSet.

• __len__ –

  Returns the number of threads contained in this ThreadSet.

__init__

__init__(*threads: Thread)

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 classmethod

with_args(*args: Args) -> Callable[[Callable], ThreadSet]

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

spam.start_and_join()

repeat classmethod

repeat(times: int) -> Callable[[Callable], ThreadSet]

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

workers.start_and_join()

target classmethod

target(target: Callable[P, None]) -> Callable[P, Thread]

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

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

start

start()

Start the threads in this ThreadSet.

Also see Thread.start().

join

join(timeout: float | None = None)

Join the threads in this ThreadSet.

Also see Thread.join().

Parameters:

• timeout

  (float | None, default: None ) –

  The timeout for each individual join to complete.

start_and_join

start_and_join(join_timeout: float | None = None)

Start the threads in this ThreadSet, then join them.

is_alive

is_alive() -> Iterable[bool]

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

any_is_alive

any_is_alive() -> bool

any(self.is_alive())

all_are_alive

all_are_alive() -> bool

all(self.is_alive())

all_are_not_alive

all_are_not_alive() -> bool

not self.any_is_alive()

__or__

__or__(other)

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

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

__ior__

__ior__(other)

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

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

__len__

__len__() -> int

Returns the number of threads contained in this ThreadSet.

len(ThreadSet(...))