友好的 Python：从其他语言移植

我们总说 Pythonic，我自己也会说某段代码不够 Pythonic，但什么是 Pythonic 呢？不讲清楚，我会被人说老登。为了避免如此，我决定再写写。 要说一段代码是 Pythonic 还是 Rustic、Go-istic、Java-ish 最直观的方法就是把它们放一起来对比，这个场景就是代码移植。其实，代码在特定语言里怎么写，是由语言特性决定的，但肯定有些部分，什么语言都能这么用，那么这些移植代码最终会变成什么样子，就更多是由写代码的人（也不一定是人）决定的了。下面我就从其他语言找些例子，看看如果把它们用 Python 重写，应该是什么样的。 opendal.Operator 不多介绍这个库了，不影响本文的主旨，咱直接从它的官方文档里截取一段最原汁原味的 Rust 代码： // Pick a builder and configure it. let mut builder = services::S3::default(); builder.bucket("test"); // Init an operator let op = Operator::new(builder)? // Init with logging layer enabled. .layer(LoggingLayer::default()) .finish(); // Use operator 因为没有任何 Python 不支持的语言特性，这一段可以 1:1 地翻译成 Python，简单地去掉 default 和 new 即可： builder = services.S3() builder.bucket("test") op = Operator(builder) \ .layer(LoggingLayer()) \ .finish() 我相信一个初学 Python 的人，就能写出这样的代码。但仔细咂摸，这好像不太像 Python，特别注意 operator 的构造方法。于是我们搬出 Xuanwo 著名的第一性原理，暂时忽略细节，思考这段代码本质在做什么。就像你摘掉眼镜看，如果不近视就眯缝着眼看，才会看到它的轮廓。 它其实就是在构造一个 S3 的 Operator 对象，并指定了一些参数。那么为什么需要一个 builder 对象呢？因为这些参数是可选的，而 Rust 不支持函数的可选参数。那 Python 没这限制，为啥还要搞个 builder 呢？实际上这个就是设计模式中的建造者模式（Builder Pattern），这也是它在 Python 里并不常见的原因。现在我们去掉 builder： op = Operator( service="s3", bucket="test", ) op.layer(LoggingLayer()) # Use operator 这就对味了（你怎么知道 opendal 的 Python binding 是这么写的？）。 其实 Javascript 也不支持可选参数，但它可以利用对象啊，所以如果用 Javascript 写的话，可能是这样： const op = new Operator({ service: "s3", bucket: "test", }); 因此在从 Javascript 移植到 Python 时也要注意这个区别，不能把这种对象也照搬成字典。 downloadFile 和 uploadFile 继续上面提到的 JS 语言，这门语言有一个突出的特征是喜欢用回调函数，特别是在 ES5 时代。因为 JS 是单线程，同步的耗时操作（如 I/O 操作）一旦发生阻塞，就会冻结整个执行流程，因此需要通过异步 I/O 或 worker 等技术将这些操作交由外部来处理。当任务执行完成后，结果会被重新送回事件队列。在 Promise 出现之前，程序只能通过回调函数来描述异步完成后的后续执行逻辑。比如这个例子： function downloadFile( url: string, onSuccess: (data: string) => void, onError: (err: Error) => void, onComplete: () => void ) 一个函数接受三个回调，这也是由于在 JS 中有 function，有箭头函数，定义回调相当方便和自然，那在 Python 中呢？因为缺少花括号，并没有一种原地定义函数的快捷方式，lamda 关键字又很鸡肋。虽说硬写也能写，但很丑，调一个函数要先 def 定义三个。其实，上面的函数在有 Promise 之后大概会像这样调用： downloadFile(url) .then((data) => { // onSuccess }) .catch((err) => { // onError }) .finally(() => { // onComplete }); // Or using async/await try { const data = await downloadFile(url); // onSuccess } catch (err) { // onError } finally { // onComplete } Python 可以借鉴这个风格： try: data = await download_file(url) # onSuccess except Exception as err: pass # onError finally: pass # onComplete 好，现在假如再增加一个回调 on_progress 呢？try/except/finally 都用完了，没有更多的关键字了啊，难道又把这个回调塞回函数参数吗： def download_file( url: str, on_progress: Callable[[bytes], None] ) -> bytes: ... 还有没有办法可以避免回调呢？有的，思考下，这个 onProgress 回调的作用是每当下载到一个部分(chunk)时执行一些动作。在 Python 里描述一段一段执行的，可以用生成器，在这里就是异步生成器： downloaded: bytes = b"" try: async for chunk in download_file(url): # process chunk downloaded += chunk # onSuccess except Exception as err: pass # onError finally: pass # onComplete 这就很 Pythonic 了。但是接下来思考一个问题，反过来的函数怎么写？刚才是下载读取数据，那写入上传数据呢？JS 完全可以用一样的回调写法。但 Python 里就不一样了，首先我们要思考输入参数是什么，当然我们可以提供一个生成器，在被读取时加入逻辑： async def file_chunks(): async for chunk in read_file_in_chunks("large_file"): # process chunk yield chunk try: await upload_file(file_chunks()) # onSuccess except Exception as err: pass # onError finally: pass # onComplete 但我觉得这里有一点不好，在于在生成器中还是由我们（生产者）控制了生成数据块的节奏，这里本应该由上传函数（消费者）根据现在的网络传输状况来控制才对。所以在这样的场景里，典型的做法应该是传入一个 file-like 对象，然后利用装饰器传入处理函数： class UploadFile: def __init__(self, fp): self._fp = fp self._callback = None def callback(self, func): self._callback = func return self def read(self, size=-1): chunk = self._fp.read(size) if self._callback: self._callback(chunk) return chunk with open("large_file", "rb") as f: upload_fp = UploadFile(f) @upload_fp.callback def on_progress(chunk): pass # process chunk try: await upload_file(upload_fp) # onSuccess except Exception as err: pass # onError finally: pass # onComplete 这个 UploadFile 定义只是为了更加通用，如果一次性使用也可以不用装饰器方式而是直接写在类里。这里我原地定义了一个函数传递给了 UploadFile，其实这是另一种传回调的方式，但这样无疑更加 Pythonic。 这一节的例子只覆盖了下载和上传这样的数据传输的场景，对于更一般的场景，需要调用方提供回调函数的情况，我们可以套用下面的公式： 如果只需要一个回调函数，考虑用装饰器 如果回调函数是关于成功和失败的，考虑用 try/except/finally 如果需要注册更多的回调函数，用类继承和多态 useEffect 最后我们再看看大名鼎鼎的 React 框架中的 useEffect。这个函数大家再熟悉不过了： const useEffect = (effect: () => (void | (() => void)), deps: any[]) => void 光写这个函数签名我就有点晕了，输入一个函数，返回一个函数，函数里还可能互相引入变量，各种闭包，这就是 JS，就是任性。要是写成 Python 呢？要是原封不动的话，可能是这样： def my_effect(): # effect def cleanup(): # cleanup return cleanup use_effect(my_effect, []) 虽然语言特性完全支持这么写，但这太不 Pythonic 了，我们不喜欢 nested，我们喜欢展平（谁叫展平？）。可是怎么展平呢？我们回到第一性原理考虑，这里的本质就是在事件发生时，执行某个动作，在事件结束时，执行清理动作。这个过程，熟悉 Python 的同学很快就想到了上下文管理器。消费上下文管理器是用 with 语句，而提供方定义上下文管理器最方便的是用 contextlib.contextmanager： from contextlib import contextmanager @contextmanager def my_effect(): # effect try: yield finally: # cleanup use_effect(my_effect, []) 注意到 use_effect 第一个参数是个函数，回顾上节的公式，凡是接受一个函数为参数的，都可以换成用装饰器，再把它和之前的 contextmanager 装饰器合并一下： @use_effect(deps=[]) def my_effect(): # effect try: yield finally: pass # cleanup 可以说这样的写法就是 native 地不能再 native 的 Python 了，那真叫一个地地弟弟道道到到，你可以在简历中写自己精通 Python 了。 结语 在文中我假想了一些移植的场景，实际上我相信大家没有多少机会真的做移植这件事，再说了，并没有人要把 useEffect 改用 Python 写。这只是一种思维训练，如果同样的功能在 Python 实现并封装成库，