提问人:Vinayak 提问时间:7/6/2023 更新时间:7/7/2023 访问量:51
如何在单独的文件中访问类外部定义的变量,而不会遇到循环导入?
How do I access a variable defined outside my class in a separate file without running into circular imports?
问:
我想访问在类的实例中定义的。这是我的代码my_variable
main.py
SecondThread
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...")
如果我运行,我会得到:python3 main.py
ImportError: cannot import name 'SecondThread' from partially initialized module 'second_thread' (most likely due to a circular import) (/home/ubuntu/test/second_thread.py)
如果我删除 并在 in second_thread.py之前添加一个右键,我会得到:from main import my_variable
global my_variable
if not my_variable.is_set():
NameError: name 'my_variable' is not defined
如果我在单个 main.py 文件中定义所有线程类,而不是将它们放在单独的文件中,这将起作用,但我不想这样做。
我怎样才能避免这种情况?
答:
1赞
blhsing
7/6/2023
#1
通常,您可以通过将导入会导致循环导入问题的模块的语句放在函数中来避免循环导入,以便在有问题的模块导入当前模块时不会执行该语句。import
然而,在这种情况下,当主程序运行时,它是由解释器以名称而不是 “导入”的,所以当 时,导入系统找不到导入缓存中指定的模块,因此继续再次导入主模块,创建另一个对象,该对象不与主程序运行时创建的对象共享。__main__
main
second_thread.py
import main
main
Event
因此,要访问解释器最初加载的主模块,您需要指定名称:__main__
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...")
此外,你不必要地在打电话后打电话,所以又变成了。删除调用,程序将按预期工作。my_variable.clear()
my_variable.set()
my_variable.is_set()
False
评论