Python循环导入问题：常规import方法无效，求解决方案？

针对你遇到的观察者模式中的循环导入问题，以下是几种实用的解决方法：

方案1：使用字符串类型注解（Python 3.7+） #

Python 3.7及以上版本支持将类型注解写成字符串形式，模块加载时不会立即解析类型，而是在运行时再查找对应的类，从而避免循环加载时的未定义问题。

修改my_subject.py中的类型注解：

from abc import ABC, abstractmethod
from random import randrange
import my_observer


class Subject(ABC):
    @abstractmethod
    def attach(self, observer: "my_observer.Observer") -> None:
        pass

    @abstractmethod
    def detach(self, observer: "my_observer.Observer") -> None:
        pass

    @abstractmethod
    def notify(self) -> None:
        pass


class ConcreteSubject(Subject):
    _state: int = None
    _observers: list["my_observer.Observer"] = []

    def attach(self, observer: "my_observer.Observer") -> None:
        print("Subject: Attached an observer.")
        self._observers.append(observer)

    def detach(self, observer: "my_observer.Observer") -> None:
        self._observers.remove(observer)

    # 其余代码保持不变

修改my_observer.py中的类型注解：

from abc import ABC, abstractmethod
import my_subject


class Observer(ABC):
    @abstractmethod
    def update(self, subject: "my_subject.Subject") -> None:
        pass


class ConcreteObserverA(Observer):
    def update(self, subject: "my_subject.Subject") -> None:
        if subject._state < 3:
            print("ConcreteObserverA: Reacted to the event")


class ConcreteObserverB(Observer):
    def update(self, subject: "my_subject.Subject") -> None:
        if subject._state == 0 or subject._state >= 2:
            print("ConcreteObserverB: Reacted to the event")

方案2：将抽象基类提取到独立模块 #

把Subject和Observer这两个抽象基类抽离到单独模块，消除业务模块间的直接依赖。

新建my_abcs.py文件：

from abc import ABC, abstractmethod


class Subject(ABC):
    @abstractmethod
    def attach(self, observer: "Observer") -> None:
        pass

    @abstractmethod
    def detach(self, observer: "Observer") -> None:
        pass

    @abstractmethod
    def notify(self) -> None:
        pass


class Observer(ABC):
    @abstractmethod
    def update(self, subject: "Subject") -> None:
        pass

修改my_subject.py：

from abc import ABC, abstractmethod
from random import randrange
from my_abcs import Subject
import my_observer


class ConcreteSubject(Subject):
    _state: int = None
    _observers: list[my_observer.Observer] = []

    def attach(self, observer: my_observer.Observer) -> None:
        print("Subject: Attached an observer.")
        self._observers.append(observer)

    def detach(self, observer: my_observer.Observer) -> None:
        self._observers.remove(observer)

    def notify(self) -> None:
        print("Subject: Notifying observers...")
        for observer in self._observers:
            observer.update(self)

    def some_business_logic(self) -> None:
        print("\nSubject: I'm doing something important.")
        self._state = randrange(0, 10)

        print(f"Subject: My state has just changed to: {self._state}")
        self.notify()

修改my_observer.py：

from abc import ABC, abstractmethod
from my_abcs import Observer
import my_subject


class ConcreteObserverA(Observer):
    def update(self, subject: my_subject.Subject) -> None:
        if subject._state < 3:
            print("ConcreteObserverA: Reacted to the event")


class ConcreteObserverB(Observer):
    def update(self, subject: my_subject.Subject) -> None:
        if subject._state == 0 or subject._state >= 2:
            print("ConcreteObserverB: Reacted to the event")

方案3：局部导入（将导入放到函数内部） #

把模块级导入移到需要使用的函数/方法内部，延迟导入时机，避免模块加载时的循环依赖。

修改my_subject.py：

from abc import ABC, abstractmethod
from random import randrange


class Subject(ABC):
    @abstractmethod
    def attach(self, observer) -> None:
        pass

    @abstractmethod
    def detach(self, observer) -> None:
        pass

    @abstractmethod
    def notify(self) -> None:
        pass


class ConcreteSubject(Subject):
    _state: int = None
    _observers: list = []

    def attach(self, observer) -> None:
        import my_observer
        if isinstance(observer, my_observer.Observer):
            print("Subject: Attached an observer.")
            self._observers.append(observer)

    def detach(self, observer) -> None:
        import my_observer
        if isinstance(observer, my_observer.Observer):
            self._observers.remove(observer)

    def notify(self) -> None:
        print("Subject: Notifying observers...")
        for observer in self._observers:
            observer.update(self)

    def some_business_logic(self) -> None:
        print("\nSubject: I'm doing something important.")
        self._state = randrange(0, 10)

        print(f"Subject: My state has just changed to: {self._state}")
        self.notify()

修改my_observer.py：

from abc import ABC, abstractmethod


class Observer(ABC):
    @abstractmethod
    def update(self, subject) -> None:
        pass


class ConcreteObserverA(Observer):
    def update(self, subject) -> None:
        import my_subject
        if isinstance(subject, my_subject.Subject) and subject._state < 3:
            print("ConcreteObserverA: Reacted to the event")


class ConcreteObserverB(Observer):
    def update(self, subject) -> None:
        import my_subject
        if isinstance(subject, my_subject.Subject) and (subject._state == 0 or subject._state >= 2):
            print("ConcreteObserverB: Reacted to the event")

以上三种方案都能解决循环导入问题，其中方案1最简单快捷，方案2更利于代码架构维护，方案3适合临时调整场景。

内容的提问来源于stack exchange，提问作者MRSMPRSS