关于 HMR 的实现
上期 谈到,我在 Python 实现了反应性原语,然后 又实现了 HMR,但是没细讲,这次写一下从反应性原语到 HMR 是怎么实现的。
我也新开了个 仓库,未来放 hmr 生态的东西(如果社区有反响的话),不过由于这算是临时起意中的临时起意(歪楼项目中的歪楼项目),所以代码还是在原仓库中没有拿出来。这个仓库目前就是一个 README 哈哈哈
如果想要直观理解 hmr 对 Python 到底意味着什么,可以看看 这个仓库 里的两个视频,另外我还分别给 FastAPI 和 Flask 开了 Discussion,分别也附上了精致的录屏,欢迎来下面讨论:
FastAPI | Flask (我还准备给 gradio、streamlit、pytest、litestar 也提一下)
不过没时间了,明天我要回家了,所以先不急着搞了
怎么实现的
正如我仓库中所说,HMR 实现可以拆成三部分:
- 实现那几个反应性原语
- 让加载进来的包,能自动跟踪依赖
- 实现一个 Reactive 的 context(按 item 追踪依赖)
- 实现一个自定义的 ModuleType,使用这个 Reactive 的 context
- 实现一个 Loader 以及一个 MetaPathFinder,并添加到
sys.meta_path中,这样新加载进来的模块就会用我们的机制来初始化
- 通过
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:...剩余内容已隐藏
关于 HMR 的实现
上期 谈到,我在 Python 实现了反应性原语,然后 又实现了 HMR,但是没细讲,这次写一下从反应性原语到 HMR 是怎么实现的。
我也新开了个 仓库,未来放 hmr 生态的东西(如果社区有反响的话),不过由于这算是临时起意中的临时起意(歪楼项目中的歪楼项目),所以代码还是在原仓库中没有拿出来。这个仓库目前就是一个 README 哈哈哈
如果想要直观理解 hmr 对 Python 到底意味着什么,可以看看 这个仓库 里的两个视频,另外我还分别给 FastAPI 和 Flask 开了 Discussion,分别也附上了精致的录屏,欢迎来下面讨论:
FastAPI | Flask (我还准备给 gradio、streamlit、pytest、litestar 也提一下)
不过没时间了,明天我要回家了,所以先不急着搞了
怎么实现的
正如我仓库中所说,HMR 实现可以拆成三部分:
- 实现那几个反应性原语
- 让加载进来的包,能自动跟踪依赖
- 实现一个 Reactive 的 context(按 item 追踪依赖)
- 实现一个自定义的 ModuleType,使用这个 Reactive 的 context
- 实现一个 Loader 以及一个 MetaPathFinder,并添加到
sys.meta_path中,这样新加载进来的模块就会用我们的机制来初始化
- 通过
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:...剩余内容已隐藏