“用过 Rust 后，我写 Python 的方法都变了！”

近年来，Rust 以其安全性出名，逐渐被各大科技巨头所拥抱——那么，其他主流语言是否可以参考 Rust 的编程思想呢？本文作者以 Python 为例，做了一番尝试。

原文链接：https://kobzol.github.io/rust/python/2023/05/20/writing-python-like-its-rust.html

未经允许，禁止转载！

作者 | Jakub Beránek

译者 | ChatGPT 责编 | 郑丽媛

出品 | CSDN（ID：CSDNnews）

从几年前开始，我尝试用 Rust 进行编程，它逐渐改变了我在其他编程语言中设计程序的方式，尤其是 Python。

在开始用 Rust 之前，我通常是以一种非常动态、不太严谨的方式来编写 Python 代码，没有类型提示，到处传递和返回字典，偶尔还回退到“字符串类型”接口。然而，在体验了 Rust 类型系统的严格性，并注意到它“通过 construction”防止的所有问题后，每当我回到 Python 时，就会突然变得相当焦虑，因为我没有得到同样的保证。

明确一点，我在这里所说的“保证”并不是指内存安全（Python 在原有情况下已相对安全），而是指“健全性”——设计很难或根本不可能被滥用的 API，从而防止未定义行为和各种错误的概念。

在 Rust 中，错误使用的接口通常会导致编译错误。而在 Python 中，这样的错误程序还是可以执行的，但如果你使用类型检查器（如 pyright）或带有类型分析器的 IDE（如 PyCharm），你就可以得到类似水平的快速反馈，以了解可能存在的问题。

最终，我开始在我的 Python 程序中采用一些来自 Rust 的概念，基本上可以归结为两点：尽可能使用类型提示，以及坚持经典的“使非法状态不可表示”原则。我试着对那些将被维护一段时间的程序以及一次性实用脚本都这样做——因为根据我的经验，后者往往会变成前者，而这种方法会让程序更易于理解和修改。

在本文中，我将展示几个将该方法应用于 Python 程序的示例。虽然这并不是什么高深的科学，但我觉得把它们记录下来可能会有用。

注意：本文中包含了很多关于编写 Python 代码的观点，我不想在每句话中都加上“在我看来”，所以请把本文中的一切都仅仅看作是我对此问题的观点，而不是试图宣传某些普遍真理。同样，我也不主张本文所提出的想法都是在 Rust 中发明的，它们在其他语言中也有使用。

类型提示

首先，最重要的是要尽可能地使用类型提示，特别是在函数签名和类属性中。当我看到一个像这样的函数签名时：

def find_item(records, check):

从函数签名本身来看，我完全无法理解其中发生了什么：它是一个列表，字典还是数据库连接？是一个布尔值还是函数？这个函数的返回值是什么？如果它失败了会发生什么？会引发异常还是返回某个值？要找到这些问题的答案，我要么去阅读函数的主体（通常还要递归地阅读它调用的其他函数的主体，这非常烦人），要么只能阅读它的文档（如果有的话）。虽然文档中可能包含了关于该函数的有用信息，但不应该必须使用文档来回答前面的问题。很多问题可以通过内置机制，即类型提示来回答。

def find_item(records: List[Item],check: Callable[[Item], bool]
) -> Optional[Item]:

写函数签名是否花费更多时间？是的。但这是个问题吗？不是，除非我的编码速度受到每分钟写入字符数量的限制，而这并不常见。明确地写出类型，迫使我思考函数实际提供的接口是什么，以及如何使其尽可能严格，让调用者难以错误地使用它。通过上面的函数签名，我可以很好地了解如何使用函数，传递什么参数，以及可以期望从函数中返回什么。此外，与文档注释不同的是，当代码发生变化时，文档注释很容易过时，而当我更改类型但未更新函数的调用者时，类型检查器会提醒我。如果我对什么感兴趣，我也可以直接使用，并立即看到该类型看起来是怎样的。

当然，我并不是绝对主义者，如果描述单个参数需要嵌套五层类型提示，我通常会放弃，并使用一个更简单但不太精确的类型。根据我的经验，这种情况不常发生，如果它真的发生了，它实际上可能预示了代码的问题——如果你的函数参数既可以是数字，又可以是字符串元组或将字符串映射为整数的字典，这可能意味着你需要重构和简化它。

使用数据类（Dataclasses）代替元组或字典

使用类型提示只是一方面，它仅描述了函数的接口是什么，第二步是尽可能准确地"锁定"这些接口。一个典型的例子是，从函数返回多个值（或单个复杂值），有一种懒惰且快速的方法是返回一个元组：

def find_person(...) -> Tuple[str, str, int]:

很好，我们知道我们要返回三个值，它们是什么？第一个字符串是这个人的名字吗？第二个字符串是姓氏吗？数字是什么？是年龄吗？还是某个列表中的位置？亦或是社会保障号码？这种类型的编码并不透明，除非你查看函数体，否则你根本不知道这代表着什么。

接下来如果要 "改进 "这一点，可以返回一个字典：

def find_person(...) -> Dict[str, Any]:...return {"name": ...,"city": ...,"age": ...}

现在，我们实际上可以知道各个返回属性是什么了，但我们又必须检查函数体才能发现。从某种意义上说，这个类型变得更糟了，因为现在我们甚至不知道各个属性的数量和类型。此外，当这个函数发生变化，返回的字典中的键被重命名或删除时，用类型检查器是不容易发现的，因此调用者通常必须经历非常繁琐的手动运行-崩溃-修改代码循环来进行更改。

正确的解决方案是，返回一个具有附加类型的命名参数的强类型对象。在 Python 中，这意味着我们需要创建一个类。我怀疑在这些情况下经常使用元组和字典，是因为相较于定义一个类（并为其命名），创建带参数的构造函数、将参数存储到字段中等要简单得多。自从 Python 3.7（以及使用 polyfill 包的更早版本）版本之后，有了一个更快捷的解决方案：.dataclasses。

@dataclasses.dataclass
class City:name: strzip_code: int@dataclasses.dataclass
class Person:name: strcity: Cityage: intdef find_person(...) -> Person:

你仍然需要为创建的类想一个名字，但除此之外，它已尽可能简洁，而且你可以得到所有属性的类型注释。

通过这个数据类，我明确了函数返回的内容。当我调用这个函数并处理返回值时，IDE 的自动完成功能会显示属性的名称和类型。听起来这可能很微不足道，但对我来说，这是一个很大的生产力优势。此外，当代码被重构、属性发生变化时，我的 IDE 和类型检查器会提醒我，并显示所有需要更改的位置，无需我执行程序。对于一些简单的重构（如属性重命名），IDE 甚至可以为我进行这些更改，此外，通过明确命名的类型，我可以建立一个词汇表（例如 Person、City），然后与其他函数和类共享。

代数数据类型

对我而言，在使用大多数主流语言时，最缺乏一项 Rust 的特性：代数数据类型（ADT）。它是一种非常强大的工具，可以明确描述代码处理的数据形状。例如，当我在 Rust 中处理数据包时，我可以明确列举所有可能接收到的数据包种类，并为每个数据包分配不同的数据（字段）：

enum Packet {Header {protocol: Protocol,size: usize},Payload {data: Vec<u8>},Trailer {data: Vec<u8>,checksum: usize}
}

通过模式匹配，我可以对各个变体作出反应，而编译器会检查我是否遗漏了任何情况：

fn handle_packet(packet: Packet) {match packet {Packet::Header { protocol, size } => ...,Packet::Payload { data } |Packet::Trailer { data, ...} => println!("{data:?}")}
}

这对于确保无效状态不可表示非常宝贵，从而避免了许多运行时错误。在静态类型语言中，ADT 特别有用，如果你想以统一方式处理一组类型，你需要一个共享的“名字”来引用它们。如果没有 ADT，通常会使用面向对象的接口或继承来实现这一点。当使用的类型集是开放式的时候，接口和虚拟方法可以解决，但当类型集是封闭的时候，并且你想确保处理所有可能的变体时，ADT 和模式匹配更加合适。

在像 Python 这样的动态类型语言中，实际上没有必要为一组类型起一个共享的名字，主要是因为在程序中使用的类型最初并不需要命名。不过使用类似 ADT 的工具仍然很有意义，例如可以创建一个联合类型：

@dataclass
class Header:protocol: Protocolsize: int@dataclass
class Payload:data: str@dataclass
class Trailer:data: strchecksum: intPacket = typing.Union[Header, Payload, Trailer]
# or `Packet = Header | Payload | Trailer` since Python 3.10

在这里，Packet 定义了一个新类型，它可以表示头部、负载或尾部数据包。但是，这些类别之间没有明确的标识符来区分它们，所以在程序中想要区分它们时，可以使用一些方法，比如使用“instanceof”运算符或模式匹配。

def handle_is_instance(packet: Packet):if isinstance(packet, Header):print("header {packet.protocol} {packet.size}")elif isinstance(packet, Payload):print("payload {packet.data}")elif isinstance(packet, Trailer):print("trailer {packet.checksum} {packet.data}")else:assert Falsedef handle_pattern_matching(packet: Packet):match packet:case Header(protocol, size): print(f"header {protocol} {size}")case Payload(data): print("payload {data}")case Trailer(data, checksum): print(f"trailer {checksum} {data}")case _: assert False

此处，我们必须在代码中必须包含一些分支逻辑，这样当函数收到意外数据时就会崩溃。而在 Rust 中，这将成为编译时错误，而不是 .assert False。

联合类型的一个好处是，它是在联合的类之外定义的。因此，该类不知道它被包含在联合中，这减少了代码的耦合度。而且，你甚至可以用相同的类创建多个不同的联合类型：

Packet = Header | Payload | Trailer
PacketWithData = Payload | Trailer

联合类型对于自动（反）序列化也非常有用。最近我发现了一个很棒的序列化库叫做 pyserde，它是基于备受推崇的 Rust serde 序列化框架开发的。除了许多其他不错的功能之外，它能利用类型注释来序列化和反序列化联合类型，而无需编写额外的代码：

import serde...
Packet = Header | Payload | Trailer@dataclass
class Data:packet: Packetserialized = serde.to_dict(Data(packet=Trailer(data="foo", checksum=42)))
# {'packet': {'Trailer': {'data': 'foo', 'checksum': 42}}}deserialized = serde.from_dict(Data, serialized)
# Data(packet=Trailer(data='foo', checksum=42))

你甚至可以选择如何将联合标签序列化，就像使用 serde 一样。我寻找类似的功能已经很久了，因为它对于序列化和反序列化联合类型非常有用。然而，在我尝试的大多数其他序列化库中，实现这一功能都相当繁琐。

举个例子，在处理机器学习模型的时候，我可以使用联合类型在单个配置文件中存储各种类型的神经网络（例如分类或分割的 CNN 模型）。我还发现，将不同版本的数据进行版本控制也非常有用，就像这样：

Config = ConfigV1 | ConfigV2 | ConfigV3

通过反序列化，我能读取所有以前版本的配置格式，从而保持向后兼容。

使用 NewType

在 Rust中，定义数据类型是很常见的，并不添加任何新行为，只是用来指定某种其他通用数据类型的领域和预期用法，例如整数。这种模式被称为“NewType”，在 Python 中也可以使用，例如：

class Database:def get_car_id(self, brand: str) -> int:def get_driver_id(self, name: str) -> int:def get_ride_info(self, car_id: int, driver_id: int) -> RideInfo:db = Database()
car_id = db.get_car_id("Mazda")
driver_id = db.get_driver_id("Stig")
info = db.get_ride_info(driver_id, car_id)

发现错误？

...

get_ride_info 函数的参数位置颠倒了。由于汽车 ID 和驾驶员 ID 都是简单的整数，因此类型是正确的，尽管从语义上来说，函数调用是错误的。

我们可以通过用“NewType”为不同种类的 ID 定义单独的类型来解决这个问题：

from typing import NewType# Define a new type called "CarId", which is internally an `int`
CarId = NewType("CarId", int)
# Ditto for "DriverId"
DriverId = NewType("DriverId", int)class Database:def get_car_id(self, brand: str) -> CarId:def get_driver_id(self, name: str) -> DriverId:def get_ride_info(self, car_id: CarId, driver_id: DriverId) -> RideInfo:db = Database()
car_id = db.get_car_id("Mazda")
driver_id = db.get_driver_id("Stig")
# Type error here -> DriverId used instead of CarId and vice-versa
info = db.get_ride_info(<error>driver_id</error>, <error>car_id</error>)

这是一个非常简单的模式，可以帮助捕捉那些难以发现的错误，尤其适合处理许多不同类型的 ID 和某些混在一起的度量指标。

使用构造函数

我很喜欢 Rust 的一点是，它没有真正意义上的构造函数。相反，人们倾向于使用普通函数来创建（最好是正确初始化的）结构体实例。在 Python 中，没有构造函数重载的概念，因此如果你需要以多种方式构造一个对象，通常会导致一个方法有很多参数，这些参数以不同的方式用于初始化，而且不能真正地一起使用。

相反，我喜欢用一个明确的名字来创建 "构造 "函数，以便清楚地了解如何构造对象以及从哪些数据中构造：

class Rectangle:@staticmethoddef from_x1x2y1y2(x1: float, ...) -> "Rectangle":@staticmethoddef from_tl_and_size(top: float, left: float, width: float, height: float) -> "Rectangle":

这样做可以使对象的构造更清晰，并且不允许用户传递无效数据，也更加清晰地表达了构造对象的意图。

用类型对不变量进行编码

用类型系统本身来编码在运行时只能追踪的不变量，是一个非常通用且强大的概念。在 Python（以及其他主流语言）中，我经常看到由一大堆可变状态组成的复杂类，导致这种混乱的原因之一是：代码试图在运行时跟踪对象的不变量。它必须考虑许多在理论上可能发生的情况，因为这些情况并没有被类型系统排除（例如“如果客户端被要求断开连接，但有人尝试向其发送消息，而 Socket 仍处于连接状态”等）

客户端

下面是一个典型的例子：

class Client:"""Rules:- Do not call `send_message` before calling `connect` and then `authenticate`.- Do not call `connect` or `authenticate` multiple times.- Do not call `close` without calling `connect`.- Do not call any method after calling `close`."""def __init__(self, address: str):def connect(self):def authenticate(self, password: str):def send_message(self, msg: str):def close(self):

……很简单，对吧？你只需要仔细阅读文档，并确保永远不会违反提到的规则（以免引发未定义行为或崩溃）。另一种方法是在类中填入各种断言，在运行时检查所有提到的规则，这将导致混乱的代码、遗漏的边缘情况和出错时较慢的反馈（编译时与运行时之间的区别）。问题的核心在于客户端可以存在于各种（互斥的）状态中，但它们并没有被分别建模成单独的类型，而是全部合并到一个类型中。

让我们看看，是否可以通过将不同状态拆分为单独的类型来改进这一点。

（1）首先，一个没有连接到任何东西的客户端是否有意义？似乎没有。在调用之前，这样一个没有连接的客户端无法执行任何操作。那为什么要允许这种状态存在呢？我们可以创建一个构造函数，它将返回一个连接的客户端：Clientconnectconnect。

def connect(address: str) -> Optional[ConnectedClient]:passclass ConnectedClient:def authenticate(...):def send_message(...):def close(...):

如果函数成功，它将返回一个遵守“已连接”不变式的客户端，你也不能再调用它来搞乱事情。如果连接失败，该函数可引发异常或返回一些显式错误。

（2）类似的方法也可用于状态。我们可以引入另一个类型，它拥有客户端既连接又认证的不变性：authenticated。

class ConnectedClient:def authenticate(...) -> Optional["AuthenticatedClient"]:class AuthenticatedClient:def send_message(...):def close(...):

只有当我们真正有了一个实例，我们才能开始发送消息。

（3）最后一个问题是方法。在 Rust 中（得益于破坏性移动语义），我们能够表达这样一个事实：当方法被调用时，你不能再使用客户端。但这在 Python 中是不可能的，所以我们必须使用一些变通办法。有一个解决方案是回退到运行时跟踪，在客户端引入一个布尔属性，并断言它还没有被关闭。另一种方法是完全删除该方法，只将客户端作为一个上下文管理器：

with connect(...) as client:client.send_message("foo")
# Here the client is closed

由于没有可用的方法，你无法意外地关闭客户端两次。

强类型的边界框

目标检测是一项我有时会参与的计算机视觉任务，其中程序必须在图像中检测一组边界框。边界框基本上是带有一些附加数据的矩形，在实现目标检测时，它们随处可见。不过边界框有一个令人讨厌的问题是：有时它们是规范化的（矩形的坐标和大小在区间内），但有时它们是非规范化的（坐标和大小受其所附图像的尺寸限制）。当你通过许多数据预处理或后处理的函数发送边界框时，很容易混淆这一点，例如多次规范化边界框，这就会导致非常麻烦的调试错误。

这种情况发生过好几次，所以我决定：将这两种类型的边界框拆分为两个单独的类型，以此来有效解决问题：NormalizedBoundingBox 和 DenormalizedBoundingBox。

@dataclass
class NormalizedBBox:left: floattop: floatwidth: floatheight: float@dataclass
class DenormalizedBBox:left: floattop: floatwidth: floatheight: float

这样分离之后，规范化和非规范化的边界框就不容易混淆了。不过我们还可以再做一些改进，把代码变得更符合“人体工学”。

（1）通过组合或继承来减少重复：

@dataclass
class BBoxBase:left: floattop: floatwidth: floatheight: float# Composition
class NormalizedBBox:bbox: BBoxBaseclass DenormalizedBBox:bbox: BBoxBaseBbox = Union[NormalizedBBox, DenormalizedBBox]# Inheritance
class NormalizedBBox(BBoxBase):
class DenormalizedBBox(BBoxBase):

（2）添加一个运行时检查，以确保边界框确实是规范化的：

class NormalizedBBox(BboxBase):def __post_init__(self):assert 0.0 <= self.left <= 1.0...

（3）添加一个在两种表示之间进行转换的方法。在某些情况下，我们可能想要知道明确的表示形式，但有时候我们也希望能使用通用接口（“任何类型的边界框”）进行操作。在这种情况下，我们应该能够将“任何边界框”转换为其中一种表示形式：

class BBoxBase:def as_normalized(self, size: Size) -> "NormalizeBBox":def as_denormalized(self, size: Size) -> "DenormalizedBBox":class NormalizedBBox(BBoxBase):def as_normalized(self, size: Size) -> "NormalizedBBox":return selfdef as_denormalized(self, size: Size) -> "DenormalizedBBox":return self.denormalize(size)class DenormalizedBBox(BBoxBase):def as_normalized(self, size: Size) -> "NormalizedBBox":return self.normalize(size)def as_denormalized(self, size: Size) -> "DenormalizedBBox":return self

通过这个接口，我可以兼顾正确性和人性化的统一界面。

注意：如果你想给父类/基类添加一些共享方法，返回对应类的实例，你可以在 Python 3.11 中使用 typing.Self。

class BBoxBase:def move(self, x: float, y: float) -> typing.Self: ...class NormalizedBBox(BBoxBase):...bbox = NormalizedBBox(...)
# The type of `bbox2` is `NormalizedBBox`, not just `BBoxBase`
bbox2 = bbox.move(1, 2)

更安全的互斥锁

在 Rust 中，互斥锁通常通过一个非常好的接口提供，这有两个好处：

（1）当你锁定互斥锁时，会得到一个“守卫”对象，该对象在销毁时可自动解锁互斥锁，主要利用了可靠的 RAII 机制：

{let guard = mutex.lock(); // locked here...
} // automatically unlocked here

这意味着，不会出现忘记解锁互斥锁的情况。在 C++ 中也有类似机制如 std::mutex，但它提供了一种没有“守卫”对象的显式/接口，这意味着其仍可能被错误使用。

（2）受互斥锁保护的数据直接存储在互斥锁（结构体）中，通过这种设计，无法在没有锁定互斥锁的情况下访问受保护的数据。你必须先锁定互斥锁以获取“守卫”对象，然后再访问数据：

let lock = Mutex::new(41); // Create a mutex that stores the data inside
let guard = lock.lock().unwrap(); // Acquire guard
*guard += 1; // Modify the data using the guard

这与主流语言（包括 Python）中常见的互斥锁 API 完全不同——在主流语言中，互斥锁和受其保护的数据是分开的，因此在访问数据之前很容易忘记锁定互斥锁：

mutex = Lock()def thread_fn(data):# Acquire mutex. There is no link to the protected variable.mutex.acquire()data.append(1)mutex.release()data = []
t = Thread(target=thread_fn, args=(data,))
t.start()# Here we can access the data without locking the mutex.
data.append(2)  # Oops

虽然在 Python 中，我们无法获得与 Rust 完全相同的功能，但它也不是一无是处。Python 锁实现了上下文管理器接口，这意味着你可以在代码块中使用它们，确保它们在作用域结束时自动解锁，甚至我们还可以更进一步：使用 with 语句。

import contextlib
from threading import Lock
from typing import ContextManager, Generic, TypeVarT = TypeVar("T")# Make the Mutex generic over the value it stores.
# In this way we can get proper typing from the `lock` method.
class Mutex(Generic[T]):# Store the protected value inside the mutex def __init__(self, value: T):# Name it with two underscores to make it a bit harder to accidentally# access the value from the outside.self.__value = valueself.__lock = Lock()# Provide a context manager `lock` method, which locks the mutex,# provides the protected value, and then unlocks the mutex when the# context manager ends.@contextlib.contextmanagerdef lock(self) -> ContextManager[T]:self.__lock.acquire()try:yield self.__valuefinally:self.__lock.release()# Create a mutex wrapping the data
mutex = Mutex([])# Lock the mutex for the scope of the `with` block
with mutex.lock() as value:# value is typed as `list` herevalue.append(1)

使用这种设计，只有在锁定互斥锁之后，你才能访问受保护的数据。显然，这仍是 Python，如果你是故意的，不变量仍可以被破坏——但这个方法已使得在 Python 中使用互斥锁接口更加安全。

总之，我确信在我的 Python 代码中还有更多的 "健全性模式"，但以上是我目前能想到的全部。如果你也有一些类似想法的例子或意见，欢迎留言告诉我。

推荐阅读：

▶FBI 花 3 年暴力破解 iPhone X 密码，竟成一场空？法院：搜查令已过期，证据无效

▶Java 17 采用率增长 430%、Java 11 稳居第一，最新 Java 编程语言报告来了！

▶Rust 社区管理再起“内讧”，外部专家遭排挤，核心成员主动请辞，立即生效！