如何在Python中设计支持生命周期钩子的业务基类-Python教程-PHP中文网

直接在__init__中调用钩子会导致子类重写的钩子未执行或访问不到子类属性，因父类__init__先于子类执行；应推迟至子类初始化完成后显式触发，如通过_post_init()统一调用。

如何在python中设计支持生命周期钩子的业务基类

为什么直接在 `init` 里调用钩子会出问题

很多同学一上来就在基类 __init__ 末尾加 self.on_created()，结果发现子类重写的 on_created 没执行，或者执行时访问不到子类初始化的属性。这是因为 Python 的 __init__ 是自上而下执行的：父类 __init__ 先跑完，子类自己的初始化逻辑（比如赋值 self.data）还没开始。

解决办法是把钩子调用推迟到子类 __init__ 完成之后——最稳妥的方式是让子类显式调用一个“完成初始化”的方法，或者用 __new__ + 标志位控制，但更常见、更可控的做法是：不自动触发，而是由使用者明确调用生命周期方法。

基类只定义钩子方法签名（如 on_created、on_destroyed），全部设为 pass 或抛 NotImplementedError
强制子类选择性重写，且在自己 __init__ 最后手动调用 self._post_init()，再由该方法统一触发钩子
避免在 __init__ 中直接调用可被重写的实例方法

`on_destroyed` 钩子不能依赖 `del`

__del__ 不可靠：它不保证何时执行，甚至可能在解释器关闭时才调，此时模块已卸载，引用的函数或全局对象可能为 None。更糟的是，如果对象参与循环引用，__del__ 可能根本不触发。

真正可用的方式是提供显式的销毁接口，比如 close() 或 teardown()，并在文档中强调“必须调用”。基类可内置状态检查：

python-parallelization

将顺序 Python 代码转换为并行/并发实现，用于在需要并行化 Python 代码、提升性能时使用。

下载

class BusinessBase:
    def __init__(self):
        self._is_closed = False

    def close(self):
        if self._is_closed:
            return
        self.on_destroyed()
        self._is_closed = True

    def on_destroyed(self):
        pass

每次 close() 前检查 _is_closed，防止重复清理导致异常
不把清理逻辑塞进 __del__，哪怕加了 atexit 注册也不如显式调用稳定
若业务场景涉及上下文管理，直接支持 with 语句（实现 __enter__/__exit__），比靠 __del__ 更可测

钩子参数要不要传上下文？看是否跨层级共享状态

简单业务基类通常不需要传参：钩子只是通知“我建好了”或“我要关了”，子类自己知道要干什么。但如果你的基类被多层继承（比如 DBService ← UserService ← AdminService），且某次 on_created 需要读取配置来源或请求 ID，硬编码获取就容易耦合。

这时可以约定一个轻量上下文对象（比如 dict 或空类实例），在初始化入口统一传入，并透传给所有钩子：

class BusinessBase:
    def __init__(self, context=None):
        self.context = context or {}
        self._post_init()

    def _post_init(self):
        self.on_created(self.context)

    def on_created(self, context):
        pass

传 context 而不是零散参数，便于后续扩展字段而不改函数签名
默认为空 dict，避免子类每次都要构造；非字典类型也兼容（只要支持 key 访问）
不推荐把整个实例（self）当上下文传——钩子本就是实例方法，已有 self

异步钩子怎么和同步基类共存

Python 基类本身不分同步异步，但一旦子类需要 async on_created，你就不能在普通 __init__ 里 await 它——因为 __init__ 必须是同步的。强行用 asyncio.run 会破坏事件循环，尤其在已有 loop 的环境中直接报错 RuntimeError: asyncio.run() cannot be called from a running event loop。

解法是分两条路径：同步基类保持原样，异步需求走单独的异步初始化方法：

class AsyncBusinessBase(BusinessBase):
    async def init_async(self):
        await self.on_created_async()
        return self

    async def on_created_async(self):
        pass