Flask应用中未导入模块的订阅逻辑执行问题求解

在Flask应用中，controllers.py文件定义了单例的SomeProcess实例：

class SomeProcess:
    def __init__(self):
        self.subscribers = []

    def subscribe(self, subscriber):
        self.subscribers.append(subscriber)
        return subscriber

    def notify(self, data):
        for subscriber in self.subscribers:
            subscriber(data)

some_process = SomeProcess() # 希望它是全应用唯一实例

在另一个模块notifications.py中，通过装饰器订阅该进程：

from controllers import some_process

@some_process.subscribe
def create_notification(data):
    pass

核心问题：由于从未在应用中导入notifications.py，模块未被执行，订阅操作无法完成。不想用集中导入的方式（会被代码检查工具标记、易误删），也不想设置集中订阅的位置。

方案1：配置文件驱动的自动发现 #

在项目根目录创建subscribers.cfg配置文件，统一管理需要加载的订阅模块：

[subscribers]
modules =
    notifications
    other_subscriber_module

修改controllers.py中的SomeProcess类，在初始化时读取配置并动态导入模块：

import configparser
import importlib

class SomeProcess:
    def __init__(self):
        self.subscribers = []
        self._load_subscribers_from_config()

    def _load_subscribers_from_config(self):
        config = configparser.ConfigParser()
        config.read('subscribers.cfg')
        module_names = [name.strip() for name in config.get('subscribers', 'modules').split('\n') if name.strip()]
        for name in module_names:
            importlib.import_module(name)

    # 原有subscribe、notify方法...

some_process = SomeProcess()

后续新增订阅模块，只需在配置文件中添加模块名即可，无需修改业务代码，也避免了裸导入的问题。

方案2：Flask专属的目录扫描加载 #

把所有订阅模块放到统一的subscribers目录下，每个模块暴露初始化函数，在应用启动时自动扫描加载：

1. 整理订阅模块结构：

# subscribers/notifications.py
from controllers import some_process

def init_subscriber():
    @some_process.subscribe
    def create_notification(data):
        pass

2. 在应用启动文件（如app.py）中添加扫描逻辑：

import os
import importlib
from flask import Flask

app = Flask(__name__)

def load_all_subscribers():
    subscribers_dir = os.path.join(os.path.dirname(__file__), 'subscribers')
    # 遍历目录下所有py文件（排除__init__.py）
    for filename in os.listdir(subscribers_dir):
        if filename.endswith('.py') and filename != '__init__.py':
            module_name = f'subscribers.{filename[:-3]}'
            module = importlib.import_module(module_name)
            # 执行初始化函数完成订阅
            if hasattr(module, 'init_subscriber'):
                module.init_subscriber()

load_all_subscribers()

# 其他应用配置与路由...

这种方式把订阅模块归类管理，自动加载，扩展性强，也符合Flask的项目结构习惯。

方案3：带忽略标记的显式导入（极简方案） #

在应用入口脚本（如run.py）中显式导入订阅模块，同时添加代码检查工具的忽略标记，搭配清晰注释防止误删：

from flask import Flask
from controllers import some_process

# 【重要】加载订阅模块，删除会导致通知订阅失效
import notifications  # noqa: F401
import other_subscriber  # noqa: F401

app = Flask(__name__)

# 启动逻辑...

# noqa: F401是告诉Pylint等工具忽略“导入未使用”的警告，这种方式简单直接，适合小型项目快速解决问题。

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