六、使用类型编码不变量
使用类型系统本身来编码只能在运行时跟踪的不变量是一个非常通用和强大的概念。在
Python(以及其他主流语言)中,我经常看到类是可变状态的毛茸茸的大球。这种混乱的根源之一是试图在运行时跟踪对象不变量的代码。它必须考虑理论上可能发生的许多情况,因为类型系统并没有使它们成为不可能(“如果客户端已被要求断开连接,现在有人试图向它发送消息,但套接字仍然是连接”等)。
1.Client
这是一个典型的例子:
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):
……容易吧?你只需要仔细阅读文档,并确保你永远不会违反上述规则(以免调用未定义的行为或崩溃)。另一种方法是用各种断言填充类,这些断言会在运行时检查所有提到的规则,这会导致代码混乱、遗漏边缘情况以及出现错误时反馈速度较慢(编译时与运行时)。问题的核心是客户端可以存在于各种(互斥的)状态中,但不是单独对这些状态进行建模,而是将它们全部合并为一个类型。
让我们看看是否可以通过将各种状态拆分为单独的类型4来改进这一点。
首先,拥有一个Client不与任何东西相连的东西是否有意义?好像不是这样。这样一个未连接的客户端在您无论如何调用之前无法执行任何操作connect 。那么为什么要允许这种状态存在呢?我们可以创建一个调用的构造函数 connect,它将返回一个连接的客户端:
def connect(address: str) -> Optional[ConnectedClient]:
pass
class ConnectedClient:
def authenticate(...):
def send_message(...):
def close(...):
如果该函数成功,它将返回一个支持“已连接”不变量的客户端,并且你不能connect再次调用它来搞砸事情。如果连接失败,该函数可以引发异常或返回None或一些显式错误。
类似的方法可以用于状态authenticated。我们可以引入另一种类型,它保持客户端已连接并已通过身份验证的不变性:
class ConnectedClient:
def authenticate(...) -> Optional["AuthenticatedClient"]:
class AuthenticatedClient:
def send_message(...):
def close(...):
只有当我们真正拥有an的实例后AuthenticatedClient,我们才能真正开始发送消息。
最后一个问题是方法close。在 Rust 中(由于 破坏性移动语义),我们能够表达这样一个事实,即当close调用方法时,您不能再使用客户端。这在 Python 中是不可能的,所以我们必须使用一些变通方法。一种解决方案可能是回退到运行时跟踪,在客户端中引入布尔属性,并断言close它send_message尚未关闭。另一种方法可能是close完全删除该方法并仅将客户端用作上下文管理器:
with connect(...) as client:
client.send_message("foo")
# Here the client is closed
没有close可用的方法,你不能意外关闭客户端两次。
2.强类型边界框
对象检测是我有时从事的一项计算机视觉任务,其中程序必须检测图像中的一组边界框。边界框基本上是带有一些附加数据的美化矩形,当你实现对象检测时,它们无处不在。关于它们的一个恼人的事情是有时它们被规范化(矩形的坐标和大小在interval中[0.0, 1.0]),但有时它们被非规范化(坐标和大小受它们所附图像的尺寸限制)。当你通过许多处理数据预处理或后处理的函数发送边界框时,很容易把它搞砸,例如两次规范化边界框,这会导致调试起来非常烦人的错误。
这在我身上发生过几次,所以有一次我决定通过将这两种类型的bbox分成两种不同的类型来彻底解决这个问题:
@dataclass
class NormalizedBBox:
left: float
top: float
width: float
height: float
@dataclass
class DenormalizedBBox:
left: float
top: float
width: float
height: float
通过这种分离,规范化和非规范化的边界框不再容易混合在一起,这主要解决了问题。但是,我们可以进行一些改进以使代码更符合人体工程学:
通过组合或继承减少重复:
@dataclass
class BBoxBase:
left: float
top: float
width: float
height: float
# Composition
class NormalizedBBox:
bbox: BBoxBase
class DenormalizedBBox:
bbox: BBoxBase
Bbox = Union[NormalizedBBox, DenormalizedBBox]
# Inheritance
class NormalizedBBox(BBoxBase):
class DenormalizedBBox(BBoxBase):
添加运行时检查以确保规范化的边界框实际上是规范化的:
class NormalizedBBox(BboxBase):
def __post_init__(self):
assert 0.0 <= self.left <= 1.0
...
添加一种在两种表示之间进行转换的方法。在某些地方,我们可能想知道显式表示,但在其他地方,我们想使用通用接口(“任何类型的 BBox”)。在那种情况下,我们应该能够将“任何 BBox”转换为以下两种表示之一:
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 self
def 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
有了这个界面,我可以两全其美——为了正确性而分开的类型,以及为了人体工程学而使用统一的界面。
注意:如果你想向返回相应类实例的父类/基类添加一些共享方法,你可以typing.Self从Python 3.11 开始使用:
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)
3.更安全的互斥锁
Rust中的互斥锁和锁通常在一个非常漂亮的接口后面提供,有两个好处:
当你锁定互斥量时,你会得到一个保护对象,它会在互斥量被销毁时自动解锁,利用古老的RAII机制:
{
let guard = mutex.lock(); // locked here
...
} // automatically unlocked here
这意味着你不会意外地忘记解锁互斥体。C++ 中也常用非常相似的机制,尽管不带保护对象的显式lock/unlock接口也可用于std::mutex,这意味着它们仍然可以被错误使用。
受互斥量保护的数据直接存储在互斥量(结构)中。使用这种设计,如果不实际锁定互斥体就不可能访问受保护的数据。您必须先锁定互斥量才能获得守卫,然后使用守卫本身访问数据:
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, TypeVar
T = 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 = value
self.__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.contextmanager
def lock(self) -> ContextManager[T]:
self.__lock.acquire()
try:
yield self.__value
finally:
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` here
value.append(1)
使用这种设计,你只能在实际锁定互斥锁后才能访问受保护的数据。显然,这仍然是Python,因此你仍然可以打破不变量——例如,通过在互斥量之外存储另一个指向受保护数据的指针。但是除非你的行为是敌对的,否则这会使Python中的互斥接口使用起来更安全。
不管怎样,我确信我在我的Python代码中使用了更多的“稳健模式”,但目前我能想到的就是这些。如果你有类似想法的一些示例或任何其他评论,请告诉我。
公平地说,如果你使用某种结构化格式(如 reStructuredText),文档注释中的参数类型描述可能也是如此。在那种情况下,类型检查器可能会使用它并在类型不匹配时警告你。但是,如果你无论如何都使用类型检查器,我认为最好利用“本机”机制来指定类型——类型提示。
aka discriminated/tagged unions, sum types, sealed classes, etc.
是的,除了这里描述的,新类型还有其他用例,别再对我大喊大叫了。
这被称为typestate 模式。
除非你努力尝试,例如手动调用魔术__exit__方法。