How do I access a variable defined outside my class in a separate file without running into circular imports?

Question

I want to access my_variable defined in main.py within an instance of the SecondThread class. Here's my code

main.py

import threading
import time
import signal
import traceback

from second_thread import SecondThread
from first_thread import FirstThread

my_variable = threading.Event()

def stop_handler(signum, frame):
    print(f"Signal received: {signum}")
    print(f"Stack frame:\n{traceback.print_stack(frame)}")
    global my_variable
    print("Setting my_variable...")
    my_variable.set()
    print(f"my_variable[{threading.get_native_id()}]:: {my_variable}")
    time.sleep(5)
    print("Unsetting my_variable...")
    my_variable.clear()
    print(f"my_variable[{threading.get_native_id()}]:: {my_variable}")

def start():
    signal.signal(signal.SIGTERM, stop_handler)
    signal.signal(signal.SIGINT, stop_handler)

    print(f"my_variable[{threading.get_native_id()}]: {my_variable}")

    first_thread = threading.Thread(target=FirstThread().run, daemon=True, name='FirstThread')
    first_thread.start()

    second_thread = threading.Thread(target=SecondThread().run, daemon=True, name='SecondThread')
    second_thread.start()

    while True:
        first_thread.join(1)

if __name__ == "__main__":
    start()

first_thread.py

import threading
import time

from second_thread import SecondThread

class FirstThread:
    def __init__(self):
        print(f"FirstThread[{threading.get_native_id()}]: Started")

    def run(self):
        for x in range(100):
            if 'SecondThread' in [t.name for t in threading.enumerate()]:
                print(f"{x}/100 FirstThread[{threading.get_native_id()}]: Sleeping for 10 seconds")
            else:
                print(f"{x}/100 FirstThread[{threading.get_native_id()}]: SecondThread is not running yet...")
                print(f"{x}/100 FirstThread[{threading.get_native_id()}]: Waiting 5 seconds and checking again...")
                time.sleep(5)
                if not 'SecondThread' in [t.name for t in threading.enumerate()]:
                    print(f"{x}/100 FirstThread[{threading.get_native_id()}]: Starting SecondThread again...")
                    second_thread = threading.Thread(target=SecondThread().run, daemon=True, name='SecondThread')
                    second_thread.start()
                else:
                    print(f"{x}/100 FirstThread[{threading.get_native_id()}]: SecondThread instance was found!")

            time.sleep(10)
        print(f"FirstThread[{threading.get_native_id()}]: Exiting...")

second_thread.py

import threading
import time

from main import my_variable

class SecondThread:
    def __init__(self):
        print(f"SecondThread[{threading.get_native_id()}]: Started")

    def run(self):
        for x in range(100):
            if not my_variable.is_set():
                print(f"{x}/100 SecondThread[{threading.get_native_id()}]: Sleeping for 10 seconds")
                time.sleep(10)
        print(f"SecondThread[{threading.get_native_id()}]: Exiting...")

If I run python3 main.py I get:

ImportError: cannot import name 'SecondThread' from partially initialized module 'second_thread' (most likely due to a circular import) (/home/ubuntu/test/second_thread.py)

If I remove the from main import my_variable and instead add a global my_variable right before the if not my_variable.is_set(): in second_thread.py I get:

NameError: name 'my_variable' is not defined

This works if I define all my thread classes in a single main.py file instead of placing them in separate files but I don't want to do that.

How can I avoid that?

Answer 1

Normally you can avoid circular import by placing the import statement that imports the module that would cause a circular import issue inside a function instead so that the statement does not get executed when the current module is imported by the offending module.

In this case, however, when the main program runs, it is "imported" by the interpreter under the name __main__ rather than main, so when second_thread.py does an import main, the import system does not find the module named main in the import cache, and therefore proceeds to import the main module again, creating another Event object that is not shared with the one created when the main program runs.

To access the main module initially loaded by the interpreter, you would therefore need to specify the name __main__ instead:

second_thread.py

import threading
import time
import sys    

class SecondThread:
    def __init__(self):
        print(f"SecondThread[{threading.get_native_id()}]: Started")

    def run(self):
        from __main__ import my_variable

        for x in range(100):
            if not my_variable.is_set():
                print(f"{x}/100 SecondThread[{threading.get_native_id()}]: Sleeping for 10 seconds")
                time.sleep(10)
        print(f"SecondThread[{threading.get_native_id()}]: Exiting...")

Furthermore, you unnecessarily call my_variable.clear() after calling my_variable.set(), so my_variable.is_set() becomes False again. Remove the call and the program would work as intended.

Demo: https://replit.com/@blhsing/BitterAdmirableStructs