Muspi Merol

Muspi Merol 的个人主页

马上订阅 Muspi Merol RSS 更新: https://muspimerol.site/feed

关于 HMR 的实现

2025年1月10日 20:09

上期 谈到,我在 Python 实现了反应性原语,然后 又实现了 HMR,但是没细讲,这次写一下从反应性原语到 HMR 是怎么实现的。

我也新开了个 仓库,未来放 hmr 生态的东西(如果社区有反响的话),不过由于这算是临时起意中的临时起意(歪楼项目中的歪楼项目),所以代码还是在原仓库中没有拿出来。这个仓库目前就是一个 README 哈哈哈

如果想要直观理解 hmr 对 Python 到底意味着什么,可以看看 这个仓库 里的两个视频,另外我还分别给 FastAPI 和 Flask 开了 Discussion,分别也附上了精致的录屏,欢迎来下面讨论:

FastAPI | Flask (我还准备给 gradio、streamlit、pytest、litestar 也提一下)

不过没时间了,明天我要回家了,所以先不急着搞了

怎么实现的

正如我仓库中所说,HMR 实现可以拆成三部分:

  1. 实现那几个反应性原语
  2. 让加载进来的包,能自动跟踪依赖
    1. 实现一个 Reactive 的 context(按 item 追踪依赖)
    2. 实现一个自定义的 ModuleType,使用这个 Reactive 的 context
    3. 实现一个 Loader 以及一个 MetaPathFinder,并添加到sys.meta_path中,这样新加载进来的模块就会用我们的机制来初始化
  1. 通过watchfiles监听 FS 变化,并 invalidate 对应的 module

其中第 3 部分正如其字面意思,很简单,就是 watchfiles 的基本用法,没什么创意。本文重点讲第二部分

Reactive namespace

我们需要实现一个 MutableMapping,类似 Vue 的 reactive,每个__getitem__会触发对应 item 的 track,而__setitem____delitem__会触发对应 item 的 notify。实现如下:

class Reactive[K, V](Subscribable, MutableMapping[K, V]):
    UNSET: V = object()  # type: ignore

    def __hash__(self):
        return id(self)

    def _null(self):
        return Signal(self.UNSET, self._check_equality)

    def __init__(self, initial: Mapping[K, V] | None = None, check_equality=True):
        super().__init__()
        self._signals = defaultdict[K, Signal[V]](self._null) if initial is None else defaultdict(self._null, {k: Signal(v, check_equality) for k, v in initial.items()})
        self._check_equality = check_equality

    def __getitem__(self, key: K):
        value = self._signals[key].get()
        if value is self.UNSET:
            raise KeyError(key)
        return value

    def __setitem__(self, key: K, value: V):
        with Batch():
            self._signals[key].set(value)
            self.notify()

    def __delitem__(self, key: K):
        state = self._signals[key]
        if state.get(track=False) is self.UNSET:
            raise KeyError(key)
        with Batch():
            state.set(self.UNSET)
            self.notify()

    def __iter__(self):
        self.track()
        return iter(self._signals)

    def __len__(self):
        self.track()
        return len(self._signals)

    def __repr__(self):
        self.track()
        return repr({k:...

剩余内容已隐藏

查看完整文章以阅读更多