How to run multiple threads together in simpler code in Python?

Question

I'm trying to simplify the code of threads below:

import threading

def test1():
    print("test1")

def test2():
    print('test2')

thread1 = threading.Thread(target=test1)
thread2 = threading.Thread(target=test2)
thread1.start()
thread2.start()
thread1.join()
thread2.join()

So, I want to simplify this part of code below to:

# ...

thread1 = threading.Thread(target=test1)
thread2 = threading.Thread(target=test2)
thread1.start()
thread2.start()
thread1.join()
thread2.join()

Something like one line of code below:

# ...

threading.Threads(test1, test2).start().join()

Are there any ways to do this? and it's ok if it's not one line of code as long as it's simpler.

Answer 1

just write it yourself ...

class Threads:
    def __init__(self,*functions):
        self._functions = functions
        self._threads = []

    def start(self):
        for func in self._functions:
            thread = threading.Thread(target=func)
            self._threads.append(thread)
            thread.start()
        return self

    def join(self):
        for thread in self._threads:
            thread.join()

usage:

Threads(test1, test2).start().join()

Edit: using threadpool is more pythonic

from operator import methodcaller
from multiprocessing.pool import ThreadPool

caller = methodcaller("__call__")
with ThreadPool() as pool:
     results = pool.map(caller, [test1, test2])

Answer 2

import threading

threads_list = []
for i in range(thread_count)
    thread = threading.Thread(target=target_func, args=(args,))
    thread.start()
    threads_list.append(thread)

Answer 3

You could refactor the functions so that they are the same function but with different inputs, e.g.

def test(s):
    print(s)

Then, you can use higher level ThreadPool api. This calls the test function with each of the elements in the list ['test1', 'test2'] via different threads. So you don't need to manage details such as starting or joining the threads.

from multiprocessing.pool import ThreadPool

with ThreadPool as pool:
     results = pool.map(test, ['test1', 'test2'])