How to create context manager with threading

Question

I have been trying to work out a a context manger for threading where I use a context manager to start a threading and close it once it is done. I have basically done

import threading


def test():
    print("Is the script running in test()? - ", running_thread.is_alive())


class Threads(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self, name='Thread-test')
        self.stop_event = threading.Event()

    def stop(self):
        self.stop_event.set()

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, *args, **kwargs):
        self.stop()
        print('Force set Thread Sleeper stop_event')


print("START SCRIPT")

with Threads() as running_thread:
    test()
    print("IS alive inside context? - ", running_thread.is_alive())

print("Outside context - ", running_thread.is_alive())

print('Main Thread ends')
print("Is the script alive? - ", running_thread.is_alive())

Output

START SCRIPT
Is the script running in test()? -  False
IS alive inside context? -  False
Force set Thread Sleeper stop_event
Outside context -  False
Main Thread ends
Is the script alive? -  False

however the problem seems like no threading is running and I can't really see why it doesn't start up and closes when using context manager.

How can I easily do it and safety as well.

EDIT: A good reviewer have recommended me this, not sure if im going the correct way with it:

You're creating and starting a thread, and throwing away its reference; so you'll never have an opportunity to join. A more careful approach would have an upper limit to the number of live threads as imposed on a thread pool; and would run test() in a context manager where the exit would join on all remaining threads. If you don't have an upper limit on the thread pool and you get more product data than expected, you've basically made a fork bomb.

Answer 1

The problem is that you haven't asked your thread to do anything, so it dies too quickly for you to see it as alive. You need to make it do something by passing a target to Thread.__init__, or providing a run() method.

Modifying your example with a run() method that sleeps in a loop:

import threading
import time


def test():
    print("Is the script running in test()? - ", running_thread.is_alive())


class Threads(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self, name='Thread-test')
        self.stop_event = threading.Event()

    def stop(self):
        self.stop_event.set()
        self.join()

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, *args, **kwargs):
        self.stop()
        print('Force set Thread Sleeper stop_event')

    def run(self):
        while not self.stop_event.is_set():
            time.sleep(0.1)


print("START SCRIPT")

with Threads() as running_thread:
    test()
    print("IS alive inside context? - ", running_thread.is_alive())

print("Outside context - ", running_thread.is_alive())

print('Main Thread ends')
print("Is the script alive? - ", running_thread.is_alive())

I get

START SCRIPT
Is the script running in test()? -  True
IS alive inside context? -  True
Force set Thread Sleeper stop_event
Outside context -  False
Main Thread ends
Is the script alive? -  False

Note that for the stop to work the thread has to actually check the stop_event. You also need to call join() in your stop() method. This makes the thread that calls stop() (the main thread) wait for the thread to finish.