Thread Class Documentation

Documentation for thread.Thread.

Importing the class

Here are some ways of importing the Thread class.

import thread
thread.Thread
 
from thread import Thread

Quick Start

There are main 2 ways of initializing a thread.

On-Demand

You can create a simple thread by initializing thread.Thread and passing the target function.

def my_target(): ...
 
# Recommended way
my_thread = Thread(
  target = my_target
)
 
# OR
# Not the recommended way
my_thread = Thread(my_target)

A thread can be ran by invoking the start() method

my_thread.start()

Decoratored Function

You can decorate a function with thread.threaded which uses thread.Thread. When the decorated function is invoked, it will automatically be ran in a new thread each time and return a thread.Thread object.

import thread
 
@thread.threaded
def my_target(): ...
 
worker = my_target() # thread.Thread()

Did you know?

Decorators can take in keyword arguments that change the behavior of the thread.

import thread
 
@thread.threaded(name = 'my_thread', suppress_errors = True)
def my_target(): ...

See the full list of arguments here

Initialization

This will cover the required and optional arguments initializing a thread.

Required

target (*args, **kwargs) -> Data_Out

This should be a function which takes in any arguments and returns anything.

Any arguments and keyword arguments parsed to the target function can be parsed through args and kwargs. Data_Out will be written to Thread._returned_value and can be accessed via Thread.result or Thread.get_return_value().

target can be parsed as the first argument to Thread.__init__(), although it is recommended to use only keyword arguments.

import thread
 
thread.Thread(lambda x: x + 1)
thread.Thread(target = lambda x: x + 1)

Best Practices

While you can use a lambda function, it is best to use a normal function for your LSP/Linter to infer types.

from thread import Thread
 
worker = Thread(target = lambda x: x + 1)
worker.start()
worker.join()
 
worker.result # This will be inferred as Unknown by your LSP/Linter

from thread import Thread
 
def my_target(x: int) -> int:
  return x + 1
 
worker = Thread(target = my_target)
worker.start()
worker.join()
 
worker.result # This will be inferred as int

Optional

args Sequence[Data_In] (default: ())

This should be an iterable sequence of arguments parsed to the target function.

For example:

import thread
 
thread.Thread(target = my_target, args = ('foo', 'bar'))
thread.Thread(target = my_target, args = ['foo', 'bar'])

⚠️

Args used in Thread.__init__() cannot offer static type checking. Learn More

kwargs Mapping[str, Data_In] (default: )

This should be a mapping of kwargs passed to the target function.

For example:

import thread
 
thread.Thread(target = my_target, kwargs = {'foo': 'bar'})
thread.Thread(target = my_target, kwargs = dict(foo = 'bar'))

⚠️

Kwargs used in Thread.__init__() cannot offer static type checking. Learn More

ignore_errors Sequence[type[Exception]] (default: ())

This should be an iterable sequence of all types of exceptions to ignore. Ignored errors are not propagated or logged to Thread.errors.

The Exception class can be parsed to ignore all exceptions.

For example:

import thread
 
# This ignores all exceptions
# thread.errors = []
# hooks begin execution
thread.Thread(target = raiseRuntimeError, ignore_errors = (Exception,))
 
# This propagates the exception
# thread.errors = [RuntimeError]
# thread is killed
thread.Thread(target = raiseRuntimeError, ignore_errors = [TypeError])
 
# This ignores RuntimeError
# thread.errors = []
# hooks begin execution
thread.Thread(target = raiseRuntimeError, ignore_errors = [RuntimeError])

Ignored exceptions may cause unintended skipped code execution.

import thread
 
def my_target():
  print('Before raise')
  raise Exception('oh no!')
  print('After raise')
 
thread.Thread(target = my_target, ignore_errors = (Exception,))
thread.start()
 
# > Before raise
# Program End

suppress_errors bool (default: False)

This should be a boolean indicating whether non-ignored exceptions will be propagated.

Non-ignored exceptions will be logged to Thread.errors without propagating.

import thread
 
# Thread.errors = [RuntimeError]
# hooks begin execution
thread.Thread(target = raiseRuntimeError, suppress_errors = True)

name str (default: None)

This should be a string name of the thread.

By default, a unique name is constructed of the form "Thread-N" where N is a small decimal number, or "Thread-N (target)" where "target" is target.__name__ if the target argument is specified. ~ Python threading docs.

import thread
 
thread.Thread(target = my_target, name = 'my_thread')

See threading documentation (opens in a new tab) for more details.

daemon bool (default: False)

This should be a boolean indicating whether the thread should be a daemon thread.

See threading documentation (opens in a new tab) for more details.

*args / **kwargs (default: None) Any / Mapping[str, Any]

These overloads are parsed to threading.Thread.__init__()

See threading documentation (opens in a new tab) for more details.

Properties

Attributes

These are attributes of thread.Thread class.

result Data_Out

This is what is returned from the target function.

💡

Exceptions Raised

Raised when the thread is not initialized and cannot invoke the method. You can initialize the thread by calling `Thread.__init__()`.

status thread.ThreadStatus

This is the current status of the thread.

These Are The Possible Values

This means that the thread is idle and ready to be ran.

errors List[Exception]

This is the list of errors raised by the thread.

Only non-ignored errors are logged to this list.

import thread
 
worker = thread.Thread(target = my_target)
worker.errors

hooks List[(Data_Out) -> Any | None]

This is the list of hooks added to the thread.

Hooks will be executed in the same order as they are added to the list.

import thread
 
worker = thread.Thread(target = my_target)
worker.hooks

Methods

These are methods of thread.Thread class.

start () -> None

This starts the thread.

Simply invoke Thread.start() on a thread object.

import thread
 
worker = thread.Thread(target = my_target)
worker.start()

💡

Exceptions Raised

Raised when the thread is still running and cannot invoke the method. You can wait for the thread to terminate by calling `Thread.join()` or check the status with `Thread.status`.

is_alive () -> bool

This indicates whether the thread is still alive.

Simply invoke Thread.is_alive() on a thread object.

import thread
 
worker = thread.Thread(target = my_target)
worker.is_alive()

💡

Exceptions Raised

Raised when the thread is not initialized and cannot invoke the method. You can initialize the thread by calling `Thread.__init__()`.

add_hook ((Data_Out) -> Any | None) -> None

This adds a hook to the thread.

Hooks are invoked in the worker thread after the thread successfully completes, passing Data_Out as the first argument. If a hook errors, the exception is recorded to Thread.errors, and is only propagated in the main thread when Thread.join(), Thread.get_return_value() or Thread.result is invoked.

Simply invoke Thread.add_hook(hook) on a thread object.

import thread
 
worker = thread.Thread(target = lambda x: x+1, args = (2,))
worker.add_hook(lambda x: print(f'Worker Hook: {x}')
worker.start()
 
# > Worker Hook: 3
# Program ends

💡

Exceptions Raised

Raised when the thread is not initialized and cannot invoke the method. You can initialize the thread by calling `Thread.__init__()`.

get_return_value () -> Data_Out

This halts the current thread execution until the thread completes and returns the value returned by the target function.

Simply invoke Thread.get_return_value() on a thread object.

import thread
 
worker = thread.Thread(target = my_target)
worker.get_return_value()

💡

Exceptions Raised

Raised when the thread is not initialized and cannot invoke the method. You can initialize the thread by calling `Thread.__init__()`.

join (timeout: float = None) -> bool

This halts the current thread execution until the thread completes or exceeds the timeout.

A None value for timeout will have the same effect as passing float("inf") as a timeout. The boolean returned by Thread.join() is True if the thread completed within the timeout and False if otherwise.

Simply invoke Thread.join() on a thread object.

import thread
worker = thread.Thread(target = my_target)
worker.join(5)
worker.join()

💡

Exceptions Raised

Raised when the thread is not initialized and cannot invoke the method. You can initialize the thread by calling `Thread.__init__()`.

kill (yielding: bool = False, timeout: float = 5) -> bool

This schedules the thread to be killed.

If yielding is True, it halts the current thread execution until the thread is killed or the timeout is exceeded. Similar to Thread.join(), a None value for timeout will have the same effect as passing float("inf") as a timeout.

Simply invoke Thread.kill() on a thread object.

import thread
 
worker = thread.Thread(target = my_target)
worker.kill(True, 10)
worker.kill(False)
worker.kill()

💡

Exceptions Raised

Raised when the thread is not initialized and cannot invoke the method. You can initialize the thread by calling `Thread.__init__()`.

⚠️

This only schedules the thread to be killed, and does not immediately kill the thread.

Meaning that if your target has a long time.wait() call, it will only be killed after it moves onto the next line.

Configuration Parallel Processing