作者丨kobzol
策劃丨千山
審校丨云昭
幾年前,我開始使用Rust編程,它逐漸改變了我使用其他編程語言(尤其是Python)設計程序的方式。在我開始使用Rust之前,我通常以一種非常動態和類型松散的方式編寫Python代碼,沒有類型提示,到處傳遞和返回字典,偶爾回退到“字符串類型”接口。然而,在經歷了Rust類型系統的嚴格性,并注意到它“通過構造”防止的所有問題之后,每當我回到Python并且沒有得到相同的保證時,我突然變得非常焦慮。
需要明確的是,這里的“保證”并不是指內存安全(Python本身是合理的內存安全),而是“穩健性”——設計很難或完全不可能被濫用的API的概念,從而防止未定義的行為和各種錯誤。在Rust中,錯誤使用的接口通常會導致編譯錯誤。在Python中,您仍然可以執行此類不正確的程序,但如果您使用類型檢查器(如pyright)或帶有類型分析器的IDE(如PyCharm),您仍然可以獲得類似級別的有關可能問題的快速反饋。
最終,我開始在我的Python程序中采用Rust的一些概念。它基本上可以歸結為兩件事——盡可能多地使用類型提示,并堅持讓非法狀態無法表示的原則。我嘗試對將維護一段時間的程序和 oneshot實用程序腳本都這樣做。主要是因為根據我的經驗,后者經常變成前者:)根據我的經驗,這種方法導致程序更容易理解和更改。
在本文中,我將展示幾個應用于Python程序的此類模式示例。這不是火箭科學,但我仍然覺得記錄它們可能會有用。
注意:這篇文章包含了很多關于編寫Python代碼的觀點。我不想在每句話中都加上“恕我直言”,所以將這篇文章中的所有內容僅作為我對此事的看法,而不是試圖宣傳一些普遍的真理:)另外,我并不是說所提出的想法是所有這些都是在Rust中發明的,當然,它們也被用于其他語言。
一、ype hint
首要的是盡可能使用類型提示,特別是在函數簽名和類屬性中。當我讀到一個像這樣的函數簽名時:
def find_item(records, check):
我不知道簽名本身發生了什么。是records列表、字典還是數據庫連接?是check布爾值還是函數?這個函數返回什么?如果失敗會發生什么,它會引發異常還是返回None?為了找到這些問題的答案,我要么必須去閱讀函數體(并且經常遞歸地閱讀它調用的其他函數的函數體——這很煩人),要么閱讀它的文檔(如果有的話)。雖然文檔可能包含有關函數功能的有用信息,但沒有必要將它也用于記錄前面問題的答案。很多問題都可以通過內置機制——類型提示——來回答。
def find_item(
records: List[Item],
check: Callable[[Item], bool]
) -> Optional[Item]:
我寫簽名花了更多時間嗎?是的。那是問題嗎?不,除非我的編碼受到每分鐘寫入的字符數的瓶頸,而這并沒有真正發生。明確地寫出類型迫使我思考函數提供的實際接口是什么,以及如何使其盡可能嚴格,以使其調用者難以以錯誤的方式使用它。通過上面的簽名,我可以很好地了解如何使用該函數、將什么作為參數傳遞給它以及我期望從中返回什么。此外,與代碼更改時很容易過時的文檔注釋不同,當我更改類型并且不更新函數的調用者時,類型檢查器會對我大喊大叫。如果我對什么是Item感興趣,我可以直接使用Go to definition并立即查看該類型的外觀。
在這方面,我不是一個絕對主義者,如果需要五個嵌套類型提示來描述單個參數,我通常會放棄并給它一個更簡單但不精確的類型。根據我的經驗,這種情況不會經常發生。如果它確實發生了,它實際上可能表明代碼有問題——如果你的函數參數可以是一個數字、一個字符串元組或一個將字符串映射到整數的字典,這可能表明你可能想要重構和簡化它。
二、數據類(dataclass)而不是元組(tuple)或字典(dictionary)
使用類型提示是一回事,但這僅僅描述了函數的接口是什么。第二步實際上是使這些接口盡可能精確和“鎖定”。一個典型的例子是從一個函數返回多個值(或一個復雜的值)。懶惰而快速的方法是返回一個元組:
def find_person(...) -> Tuple[str, str, int]:
太好了,我們知道我們要返回三個值。這些是什么?第一個字符串是人的名字嗎?第二串姓氏?電話號碼是多少?是年齡嗎?在某些列表中的位置?社會安全號碼?這種輸入是不透明的,除非你查看函數體,否則你不知道這里發生了什么。
下一步“改進”這可能是返回一個字典:
def find_person(...) -> Dict[str, Any]:
...
return {
"name": ...,
"city": ...,
"age": ...
}
現在我們實際上知道各個返回的屬性是什么,但我們必須再次檢查函數體才能找出答案。從某種意義上說,類型變得更糟,因為現在我們甚至不知道各個屬性的數量和類型。此外,當這個函數發生變化并且返回的字典中的鍵被重命名或刪除時,沒有簡單的方法可以用類型檢查器找出來,因此它的調用者通常必須用非常手動和煩人的運行-崩潰-修改代碼來改變循環。
正確的解決方案是返回一個強類型對象,其命名參數具有附加類型。在Python中,這意味著我們必須創建一個類。我懷疑在這些情況下經常使用元組和字典,因為它比定義類(并為其命名)、創建帶參數的構造函數、將參數存儲到字段等容易得多。自Python 3.7 (并且更快地使用package polyfill),有一個更快的解決方案-dataclasses.
@dataclasses.dataclass
class City:
name: str
zip_code: int
@dataclasses.dataclass
class Person:
name: str
city: City
age: int
def find_person(...) -> Person:
你仍然需要為創建的類考慮一個名稱,但除此之外,它已經盡可能簡潔了,并且你可以獲得所有屬性的類型注釋。
有了這個數據類,我就有了函數返回內容的明確描述。當我調用此函數并處理返回值時,IDE自動完成功能將向我顯示其屬性的名稱和類型。這聽起來可能微不足道,但對我來說這是一個巨大的生產力優勢。此外,當代碼被重構并且屬性發生變化時,我的IDE和類型檢查器將對我大喊大叫并向我顯示所有必須更改的位置,而我根本不必執行程序。對于一些簡單的重構(例如屬性重命名),IDE甚至可以為我進行這些更改。此外,通過明確命名的類型,我可以構建術語詞匯表( Person,City),然后可以與其他函數和類共享。
三、代數數據類型
在大多數主流語言中,我可能最缺乏的Rust是代數數據類型(ADT)2。它是一個非常強大的工具,可以明確描述我的代碼正在處理的數據的形狀。例如,當我在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,這通常是使用OOP接口和/或繼承來完成的。當使用的類型集是開放式的時,接口和虛方法有它們的位置,但是當類型集是封閉的,并且你想確保你處理所有可能的變體時,ADT和模式匹配更合適。
在動態類型語言(如Python)中,實際上不需要為一組類型共享名稱,主要是因為您甚至不必一開始就為程序中使用的類型命名。但是,通過創建聯合類型,使用類似于ADT的東西仍然有用:
@dataclass
class Header:
protocol: Protocol
size: int
@dataclass
class Payload:
data: str
@dataclass
class Trailer:
data: str
checksum: int
Packet = typing.Union[Header, Payload, Trailer]
# or `Packet = Header | Payload | Trailer` since Python 3.10
Packet這里定義了一個新類型,它可以是報頭、有效載荷或尾部數據包。當我想確保只有這三個類有效時,我現在可以在程序的其余部分中使用此類型(名稱)。請注意,類沒有附加明確的“標簽”,因此當我們要區分它們時,我們必須使用eginstanceof或模式匹配:
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 False
def 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
可悲的是,在這里我們必須(或者更確切地說,應該)包括煩人的assert False分支,以便函數在接收到意外數據時崩潰。在Rust中,這將是一個編譯時錯誤。
注意:Reddit上的幾個人已經提醒我,assert False實際上在優化構建( ) 中完全優化掉了python -O ...。因此,直接引發異常會更安全。還有typing.assert_never來自Python 3.11 的,它明確地告訴類型檢查器落到這個分支應該是一個“編譯時”錯誤。
聯合類型的一個很好的屬性是它是在作為聯合一部分的類之外定義的。因此該類不知道它被包含在聯合中,這減少了代碼中的耦合。您甚至可以使用相同的類型創建多個不同的聯合:
Packet = Header | Payload | Trailer
PacketWithData = Payload | Trailer
聯合類型對于自動(反)序列化也非常有用。最近我發現了一個很棒的序列化庫,叫做pyserde,它基于古老的Rustserde序列化框架。在許多其他很酷的功能中,它能夠利用類型注釋來序列化和反序列化聯合類型,而無需任何額外代碼:
import serde
...
Packet = Header | Payload | Trailer
@dataclass
class Data:
packet: Packet
serialized = 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.我一直在尋找類似的功能,因為它對(反)序列化聯合類型非常有用。dataclasses_json但是,在我嘗試過的大多數其他序列化庫(例如或)中實現它非常煩人dacite。
例如,在使用機器學習模型時,我使用聯合將各種類型的神經網絡(例如分類或分段CNN模型)存儲在單個配置文件格式中。我還發現對不同格式的數據(在我的例子中是配置文件)進行版本化很有用,如下所示:
Config = ConfigV1 | ConfigV2 | ConfigV3
通過反序列化Config,我能夠讀取所有以前版本的配置格式,從而保持向后兼容性。
四、使用newtype
在Rust中,定義不添加任何新行為的數據類型是很常見的,但只是用于指定其他一些非常通用的數據類型(例如整數)的域和預期用途。這種模式被稱為“newtype”3,它也可以用在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 (CarId vs DriverId)或某些不應混合在一起的指標(Speed vs Lengthvs等)。Temperature
五、使用構造函數
我非常喜歡Rust的一件事是它本身沒有構造函數。相反,人們傾向于使用普通函數來創建(理想情況下正確初始化)結構實例。在Python中,沒有構造函數重載,因此如果您需要以多種方式構造一個對象,有人會導致一個__init__方法有很多參數,這些參數以不同的方式用于初始化,并且不能真正一起使用。
相反,我喜歡創建具有明確名稱的“構造”函數,這使得如何構造對象以及從哪些數據構造對象變得顯而易見:
class Rectangle:
@staticmethod
def from_x1x2y1y2(x1: float, ...) -> "Rectangle":
@staticmethod
def from_tl_and_size(top: float, left: float, width: float, height: float) -> "Rectangle":
這使得構造對象變得更加清晰,并且不允許類的用戶在構造對象時傳遞無效數據(例如通過組合y1和width)。
六、使用類型編碼不變量
使用類型系統本身來編碼只能在運行時跟蹤的不變量是一個非常通用和強大的概念。在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__方法。
原文鏈接:https://kobzol.github.io/rust/python/2023/05/20/writing-python-like-its-rust.html