状态即类型,一种设计模式Draft

编程的核心在于状态。

计算机科学家会告诉你,编程语言只是定义抽象图灵机的DFA(或状态机)的一种方式。另一方面,硬件工程师会告诉你,CPU——运行程序的物理实体——本质上是一个用晶体管编码的状态机。

编程的核心在于状态。

可以说,忽略语法差异,编程语言之间的区别在于它们如何处理状态。我们可以通过考虑状态在编译时与运行时解决的比例来对语言进行分类。

编程语言的谱系
常见编程语言在运行时/编译时谱系中的位置

最左边是完全解释型语言1,如Python和Lua,它们在运行时之前甚至不知道要执行什么指令。另一边则是强类型、编译型语言,如Rust和C++,它们在编译前需要大量关于程序的信息。

无论语言位于谱系的哪个位置,都存在权衡。非常严格的语言,需要大量关于类型、内存和生命周期的信息,通常难以构建和迭代,但在运行时非常高效。而另一端的语言则相反:易于快速编写,但伴随着显著的运行时成本。

强类型语言在一个领域无疑表现出色:错误管理。通常情况下,解释型语言的哲学是在运行后调试,看看是否抛出任何意外异常。

算法Y:解释型语言调试(简化版)

  1. 编写代码
  2. 运行代码
  3. 如果没有错误,跳至步骤6
  4. 处理错误
  5. 跳至步骤2
  6. 部署

无论你更看重开发时间还是运行时性能,我认为我们都同意程序应该是正确健壮的,能够应对错误。

我们将探讨一种设计模式,它有助于将运行时逻辑分类并转移到编译时(或代码编写时,如果使用解释型语言)。实际上,它将创建一个状态管道,利用语言的类型系统来防止非法的对象状态。别担心,静态类型并不是必需的。

让我们来看一个使用这种模式进行重构的示例。

任务

老板告诉我们,必须编写一个库,用于统计任何 URL 指向的任何文件中给定字符的出现次数。他还希望我们的代码用户能够随时下载该文件。

听起来不错。让我们创建一个类来完成这个任务。

class FileDownloadCharCounter:
    def __init__(self, url):
		# 将参数保存为实例变量
        self.url = url

    def download(self):
		# 下载内容并保存以备后用
        self.downloaded_content = requests.get(self.url).text

    def create_index(self):
		# 创建一个存储字符计数的字典
        self.index = {}
        for char in self.downloaded_content:
            self.index[char] = self.index.get(char, 0) + 1

    def get_count(self, target_char):
		# 从字典中获取计数
        return self.index.get(target_char, 0)

在这个练习中,我们假设 requests.get 永远不会失败。让我们测试一下这个类。

counter = FileDownloadCharCounter(
	"https://world.hey.com/dhh/programming-types-and-mindsets-5b8490bc"
)
counter.download()
counter.create_index()
target = "a"
n = counter.get_count("a")
print(f"{target} appears {n} times")
a appears 477 times

太好了!是时候部署了吗?别急……如果你仔细观察,我们会发现有很多未捕获的异常。如果用户这样做会发生什么?

counter = FileDownloadCharCounter(
	"https://world.hey.com/dhh/programming-types-and-mindsets-5b8490bc"
)
counter.download()
target = "a"
n = counter.get_count("a")
AttributeError: 'FileDownloadCharCounter' object has no attribute 'index'

哦不。如果用户忘记创建索引,他们会得到一个 AttributeError,这并不太有帮助。这将要求他们阅读库的源代码来找出问题所在,这不是一个很好的用户体验。让我们创建一个自解释的异常,用户可以捕获并解决它:

class IndexNotCreatedException(Exception):
	pass

class FileDownloadCharCounter:
    def __init__(self, url):
		# 其他代码 ...
        self.index = {}

# 其他方法 ...

    def get_count(self, target_char):
        if len(self.index) == 0:
            raise IndexNotCreatedException
        return self.index[target_char]

现在我们得到了一个友好的

File "example.py", line 31, in get_count
    raise IndexNotCreatedException
__main__.IndexNotCreatedException

我们可以这样处理:

try:
	count = counter.get_count('a')
except IndexNotCreatedException:
	# 恢复
	pass

还有没有其他可能出错的地方?是的:

counter.create_index()
target = "a"
n = counter.get_count("a")
  File "example.py", line 13, in create_index
    for char in self.downloaded_content:
AttributeError: 'FileDownloadCharCounter' object has no attribute 'downloaded_content'

如果库用户忘记下载文件,它会抛出另一个 AttributeError,这并不太有帮助。让我们处理一下:

class FileNotDownloadedException(Exception):
    pass

class FileDownloadCharCounter:
    def __init__(self, url):
		# ...
        self.downloaded = False

    def download(self):
		# ...
        self.downloaded = True

    def create_index(self):
        if not self.downloaded:
            raise FileNotDownloadedException
		# ...

现在,用户可以像上面那样处理这个异常。还有一个 bug;你能找到它吗?假设用户不小心这样做了:

counter = FileDownloadCharCounter(
	"https://world.hey.com/dhh/programming-types-and-mindsets-5b8490bc"
)
counter.download()
counter.create_index()
# 其他重要的事情...
counter.create_index()
# 更多重要的事情...
target = "a"
n = counter.get_count("a")
print(f"{target} appears {n} times")

现在我们得到了双倍的实际计数!

a appears 954 times

并且 没有异常。这意味着这个 bug 没有被算法 Y 捕获。再次,让我们处理一下:

class IndexNotCreatedException(Exception):
    pass

class FileDownloadCharCounter:
	# 方法...

    def create_index(self):
        if len(self.index) > 0:
            raise IndexAlreadyCreatedException
		# 更多代码...

呼!这似乎是所有与状态相关的错误。这是我们的最终代码:

class IndexNotCreatedException(Exception):
    pass

class IndexAlreadyCreatedException(Exception):
    pass

class FileNotDownloadedException(Exception):
    pass

class FileDownloadCharCounter:
    def __init__(self, url):
        self.url = url
        self.index = {}
        self.downloaded = False

    def download(self):
        self.downloaded_content = requests.get(self.url).text
        self.downloaded = True

    def create_index(self):
        if not self.downloaded:
            raise FileNotDownloadedException

        if len(self.index) == 0:
            raise IndexAlreadyCreatedException

        for char in self.downloaded_content:
            self.index[char] = self.index.get(char, 0) + 1

    def get_count(self, target_char):
        if len(self.index) == 0:
            raise IndexNotCreatedException

        return self.index.get(target_char, 0)

瞧!我们的面向对象代码……糟糕透了。对于本应只是几行简单的 Python 代码,我们有 3 个自定义异常和一堆逻辑,只是为了确保没有出错。

你可以想象,如果跟踪状态的实例变量不是 3 个,而是 30 个,可能甚至无法知道或枚举所有非法状态来抛出异常。

不幸的是,许多解释型语言往往使编写健壮的代码变得繁琐。为了帮助解决这个问题,让我们引入一个新的设计模式,我将其称为类型管道,或者如果你愿意,也可以称为Typeline

存在性 状态

通过这个技巧,在静态类型语言中,具有非法状态的代码将无法编译。在动态类型语言中,如果你没有遇到 TypeError,你将拥有一个保证合法的状态。

其工作原理如下:

  • 将类型 与状态 关联
  • 保证类型 的实例对象 的存在意味着我们处于状态

就是这么简单。现在让我们尝试重构之前的代码。考虑我们程序的有效状态:

状态 URL 已知 文件已下载 文件已索引
1
2
3

很明显,这是一个简单的线性管道。状态 1 需要一个 URL 作为输入,状态 2 需要状态 1,状态 3 需要状态 2。

如果 是对应于状态 的类型,那么构造 的实例应该只能通过 的实例来访问。让我们从状态 1 开始,它只需要一个有效的 URL。

class FileURL: # 即 T_1
	def __init__(self, url):
		# 可选地验证 URL
		self.url = url

由于我们只希望 通过 的实例来构造,让我们在 上添加一个方法,该方法构造一个具有有效状态的 (文件已下载)。

class FileURL:
	# 其他方法 ...
	def download_file(self, file_url: FileURL) -> DownloadedFile:
		file_contents = requests.get(file_url).text
		return DownloadedFile(file_contents)

class DownloadedFile:
	def __init__(self, file_contents: str):
		self.contents = file_contents

我们对 重复这一过程,它表示一个已索引的文件。

class DownloadedFile:
	# 其他方法 ...
	def index_file(self) -> IndexedFile:
		index = {}
		for char in content:
            index[char] = index.get(char, 0) + 1
		return IndexedFile(index)

class IndexedFile:
	def __init__(self, index: dict[str, int]):
		self.index = index

	def get_count(self, target_char):
		return self.index.get(target_char, 0)

现在,所有这些类只是为了演示这个想法。实际上,我们并不需要让用户访问一个已下载但未索引的文件,也不需要让用户访问一个仅仅包装了字符串的 URL 类。他们只需要能够按需下载文件并获取字符计数。

因此,我们可以删除与这些状态关联的类型,并将逻辑放入一个转换方法或构造函数中。

class FileURL:
	def __init__(self, url):
		self.url = url

	# 状态 1 -> 3
	def fetch_index(self) -> CharIndex:
		# 状态 1 -> 2
		content = requests.get(url).text
		# 状态 2 -> 3
		return CharIndex(content)

class CharIndex:
	# 状态 2 -> 3
	def __init__(self, content: str):
		index = {}
		for char in content:
            index[char] = index.get(char, 0) + 1
		self.index = index

	def get_count(self, target_char):
		return self.index.get(target_char, 0)

用法:

file = FileURL("https://world.hey.com/dhh/programming-types-and-mindsets-5b8490bc")
index = file.fetch_index()
count = index.get_count('a')

你可以看到,通过转换/构造的保证,非法状态不再是一个问题。

然而,有一些注意事项。这种模式只有在以下情况下才有效:

  1. 状态在编译/代码编写时已知(即第 行的对象 应该 具有状态
  2. 只有少数这样的状态,因为你 可能 需要为每个状态创建一个新类

一些优势

在垃圾回收语言中,一个巨大的优势是内存效率。回到我们的例子,在原始类中,下载的文件与整个对象具有相同的生命周期。这意味着,只要存在对对象的引用,整个站点的字符串就会一直存储在内存中。

counter = FileDownloadCharCounter(...)
counter.download_file()  # 为字符串分配空间
counter.create_index()   # 为索引字典分配空间
counter.get_count('a')

# 在作用域结束时释放所有内容

但我们从任务中知道,在索引创建后,我们不再需要文件内容。由于Typeline围绕数据转换构建,生命周期是明确定义的。如果不再需要某块数据,垃圾回收器可以安全地销毁它。

file = FileURL(...)
index = file.fetch_index() # 字符串分配并立即释放
count = index.get_count('a') # 索引分配
# 索引在作用域结束时释放

如果我们下载的文件有数GB大小,或者我们同时抓取数千个网站,这一点尤其有用。

结论

本文介绍了一种简单的设计模式,我在编写streamrip v2时发现它非常有用,它修复了v1中大量可能的状态错误,并显著简化了代码库。虽然我之前没有遇到过这个确切的想法,但它绝非原创。Typeline 只是一种面向对象的方式,用于编码带有一些副作用管理的 DFA(确定性有限自动机)。因此,所有关于 DFA 的理论工作在这里同样适用。

如果你在任何地方发现了这种模式,请告诉我!


  1. 我们称之为解释型的现代语言实际上是通过 JIT(即时)编译进行编译的。我们使用“解释型”这个术语来表示即使程序类型不正确,仍然可以运行。 ↩︎

✦ No LLMs were used in the ideation, research, writing, or editing of this article.