ThreadSet

ThreadSet ¶

ThreadSet(*threads: Thread)

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.

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()

Methods:

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.

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(...))

ThreadSet

ThreadSet ¶

with_args classmethod ¶

repeat classmethod ¶

target classmethod ¶