简介
在本次实验中,你将学习如何在 Python 中使用生成器(generator)自定义迭代。你还将在自定义类中实现迭代器(iterator)功能,并为流式数据源创建生成器。
在实验过程中,你将修改 structure.py
文件,并创建一个名为 follow.py
的新文件。
This tutorial is from open-source community. Access the source code
💡 本教程由 AI 辅助翻译自英文原版。如需查看原文,您可以 切换至英文原版
在本次实验中,你将学习如何在 Python 中使用生成器(generator)自定义迭代。你还将在自定义类中实现迭代器(iterator)功能,并为流式数据源创建生成器。
在实验过程中,你将修改 structure.py
文件,并创建一个名为 follow.py
的新文件。
生成器(Generator)是 Python 中一项强大的特性。它们提供了一种简单而优雅的方式来创建迭代器(Iterator)。在 Python 中,当你处理数据序列时,迭代器非常有用,因为它们允许你逐个遍历一系列的值。普通函数通常返回单个值,然后停止执行。然而,生成器不同。它们可以随着时间的推移产生一系列的值,这意味着它们能够逐步产生多个值。
生成器函数的外观与普通函数相似。但关键区别在于它们返回值的方式。生成器函数不使用 return
语句来提供单个结果,而是使用 yield
语句。yield
语句很特殊。每次执行该语句时,函数的状态会被暂停,并且 yield
关键字后面的值会返回给调用者。当再次调用生成器函数时,它会从上次暂停的地方继续执行。
让我们从创建一个简单的生成器函数开始。Python 中的内置 range()
函数不支持小数步长。因此,我们将创建一个可以生成具有小数步长的数字范围的生成器函数。
def frange(start, stop, step):
current = start
while current < stop:
yield current
current += step
## Test the generator with a for loop
for x in frange(0, 2, 0.25):
print(x, end=' ')
在这段代码中,frange
函数是一个生成器函数。它用 start
值初始化变量 current
。然后,只要 current
小于 stop
值,它就会产生 current
值,然后将 current
增加 step
值。for
循环会遍历 frange
生成器函数产生的值并将其打印出来。
你应该会看到以下输出:
0 0.25 0.5 0.75 1.0 1.25 1.5 1.75
生成器的一个重要特性是它们是可耗尽的。这意味着一旦你遍历了生成器产生的所有值,就不能再用它来产生相同的序列值了。让我们用以下代码来演示这一点:
## Create a generator object
f = frange(0, 2, 0.25)
## First iteration works fine
print("First iteration:")
for x in f:
print(x, end=' ')
print("\n")
## Second iteration produces nothing
print("Second iteration:")
for x in f:
print(x, end=' ')
print("\n")
在这段代码中,我们首先使用 frange
函数创建了一个生成器对象 f
。第一个 for
循环遍历生成器产生的所有值并将其打印出来。第一次迭代之后,生成器已经耗尽,这意味着它已经产生了所有能产生的值。因此,当我们在第二个 for
循环中再次尝试遍历它时,它不会产生任何新的值。
输出:
First iteration:
0 0.25 0.5 0.75 1.0 1.25 1.5 1.75
Second iteration:
注意,第二次迭代没有产生任何输出,因为生成器已经耗尽了。
如果你需要多次遍历相同的序列值,可以将生成器封装在一个类中。这样做的话,每次开始新的迭代时,都会创建一个全新的生成器。
class FRange:
def __init__(self, start, stop, step):
self.start = start
self.stop = stop
self.step = step
def __iter__(self):
n = self.start
while n < self.stop:
yield n
n += self.step
## Create an instance
f = FRange(0, 2, 0.25)
## We can iterate multiple times
print("First iteration:")
for x in f:
print(x, end=' ')
print("\n")
print("Second iteration:")
for x in f:
print(x, end=' ')
print("\n")
在这段代码中,我们定义了一个类 FRange
。__init__
方法初始化 start
、stop
和 step
值。__iter__
方法是 Python 类中的一个特殊方法,用于创建迭代器。在 __iter__
方法内部,我们有一个生成器,它产生值的方式与我们之前定义的 frange
函数类似。
当我们创建 FRange
类的实例 f
并多次对其进行迭代时,每次迭代都会调用 __iter__
方法,该方法会创建一个全新的生成器。因此,我们可以多次获取相同的序列值。
输出:
First iteration:
0 0.25 0.5 0.75 1.0 1.25 1.5 1.75
Second iteration:
0 0.25 0.5 0.75 1.0 1.25 1.5 1.75
这一次,我们可以多次迭代,因为每次调用 __iter__()
方法时都会创建一个全新的生成器。
既然你已经掌握了生成器的基础知识,我们将使用它们为自定义类添加迭代功能。在 Python 中,如果你想让一个类可迭代,就需要实现 __iter__()
特殊方法。一个可迭代的类允许你遍历其元素,就像你可以遍历列表或元组一样。这是一个强大的特性,它使你的自定义类更加灵活且易于使用。
__iter__()
方法__iter__()
方法是使类可迭代的关键部分。它应该返回一个迭代器对象。迭代器是一个可以被迭代(遍历)的对象。实现这一点的一个简单有效的方法是将 __iter__()
定义为一个生成器函数。生成器函数使用 yield
关键字一次产生一个值的序列。每次遇到 yield
语句时,函数会暂停并返回该值。下次调用迭代器时,函数会从上次暂停的地方继续执行。
Structure
类在本次实验的设置中,我们提供了一个基础的 Structure
类。其他类(如 Stock
)可以继承这个 Structure
类。继承是创建一个新类的方式,新类可以继承现有类的属性和方法。通过为 Structure
类添加 __iter__()
方法,我们可以使它的所有子类都可迭代。这意味着任何继承自 Structure
的类都将自动具备可遍历的能力。
structure.py
文件:cd ~/project
这个命令将当前工作目录更改为 project
目录,structure.py
文件就位于该目录中。你需要处于正确的目录才能访问和修改该文件。
Structure
类的当前实现:class Structure(metaclass=StructureMeta):
_fields = []
def __init__(self, *args):
if len(args) != len(self._fields):
raise TypeError(f'Expected {len(self._fields)} arguments')
for name, val in zip(self._fields, args):
setattr(self, '_'+name, val)
Structure
类有一个 _fields
列表,用于存储属性的名称。__init__()
方法是类的构造函数。它通过检查传入的参数数量是否等于字段数量来初始化对象的属性。如果不相等,它会引发一个 TypeError
。否则,它使用 setattr()
函数设置属性。
__iter__()
方法,按顺序产生每个属性的值:def __iter__(self):
for name in self._fields:
yield getattr(self, name)
这个 __iter__()
方法是一个生成器函数。它遍历 _fields
列表,并使用 getattr()
函数获取每个属性的值。然后,yield
关键字逐个返回这些值。
现在完整的 structure.py
文件应该如下所示:
class StructureMeta(type):
def __new__(cls, name, bases, clsdict):
fields = clsdict.get('_fields', [])
for name in fields:
clsdict[name] = property(lambda self, name=name: getattr(self, '_'+name))
return super().__new__(cls, name, bases, clsdict)
class Structure(metaclass=StructureMeta):
_fields = []
def __init__(self, *args):
if len(args) != len(self._fields):
raise TypeError(f'Expected {len(self._fields)} arguments')
for name, val in zip(self._fields, args):
setattr(self, '_'+name, val)
def __iter__(self):
for name in self._fields:
yield getattr(self, name)
这个更新后的 Structure
类现在有了 __iter__()
方法,这使得它及其子类都可迭代。
保存文件。
对 structure.py
文件进行更改后,你需要保存它,以便应用这些更改。
现在,让我们通过创建一个 Stock
实例并对其进行迭代来测试迭代功能:
python3 -c "from stock import Stock; s = Stock('GOOG', 100, 490.1); print('Iterating over Stock:'); [print(val) for val in s]"
这个命令创建了一个 Stock
类的实例,该类继承自 Structure
类。然后,它使用列表推导式对该实例进行迭代,并打印每个值。
你应该会看到如下输出:
Iterating over Stock:
GOOG
100
490.1
现在,任何继承自 Structure
的类都将自动可迭代,并且迭代将按照 _fields
列表定义的顺序产生属性值。这意味着你可以轻松地遍历 Structure
任何子类的属性,而无需为迭代编写额外的代码。
现在,我们已经让 Structure
类及其子类支持迭代了。迭代是 Python 中一个强大的概念,它允许你逐个遍历一组对象。当一个类支持迭代时,它会变得更加灵活,并且可以与许多 Python 内置特性协同工作。让我们来探索一下这种迭代支持如何在 Python 中实现许多强大的功能。
在 Python 中,有像 list()
和 tuple()
这样的内置函数。这些函数非常有用,因为它们可以接受任何可迭代对象作为输入。可迭代对象是指你可以进行遍历的对象,比如列表、元组,现在还包括我们的 Structure
类实例。由于我们的 Structure
类现在支持迭代,我们可以轻松地将其实例转换为列表或元组。
Stock
实例来尝试这些操作。Stock
类是 Structure
类的子类。在终端中运行以下命令:python3 -c "from stock import Stock; s = Stock('GOOG', 100, 490.1); print('As list:', list(s)); print('As tuple:', tuple(s))"
这个命令首先导入 Stock
类,创建一个它的实例,然后分别使用 list()
和 tuple()
函数将该实例转换为列表和元组。输出将显示该实例以列表和元组形式表示的结果:
As list: ['GOOG', 100, 490.1]
As tuple: ('GOOG', 100, 490.1)
Python 有一个非常有用的特性叫做解包(Unpacking)。解包允许你将一个可迭代对象的元素一次性分配给各个变量。由于我们的 Stock
实例是可迭代的,我们可以对其使用解包特性。
python3 -c "from stock import Stock; s = Stock('GOOG', 100, 490.1); name, shares, price = s; print(f'Name: {name}, Shares: {shares}, Price: {price}')"
在这段代码中,我们创建了一个 Stock
实例,然后将其元素解包到三个变量中:name
、shares
和 price
。接着我们打印这些变量。输出将显示这些变量的值:
Name: GOOG, Shares: 100, Price: 490.1
当一个类支持迭代时,实现比较操作会变得更加容易。比较操作用于检查两个对象是否相等。让我们为 Structure
类添加一个 __eq__()
方法来比较实例。
structure.py
文件。__eq__()
方法是 Python 中的一个特殊方法,当你使用 ==
运算符比较两个对象时会调用该方法。在 structure.py
文件的 Structure
类中添加以下代码:def __eq__(self, other):
return isinstance(other, type(self)) and tuple(self) == tuple(other)
这个方法首先使用 isinstance()
函数检查 other
对象是否与 self
是同一类的实例。然后它将 self
和 other
都转换为元组,并检查这些元组是否相等。
现在完整的 structure.py
文件应该如下所示:
class StructureMeta(type):
def __new__(cls, name, bases, clsdict):
fields = clsdict.get('_fields', [])
for name in fields:
clsdict[name] = property(lambda self, name=name: getattr(self, '_'+name))
return super().__new__(cls, name, bases, clsdict)
class Structure(metaclass=StructureMeta):
_fields = []
def __init__(self, *args):
if len(args) != len(self._fields):
raise TypeError(f'Expected {len(self._fields)} arguments')
for name, val in zip(self._fields, args):
setattr(self, '_'+name, val)
def __iter__(self):
for name in self._fields:
yield getattr(self, name)
def __eq__(self, other):
return isinstance(other, type(self)) and tuple(self) == tuple(other)
添加 __eq__()
方法后,保存 structure.py
文件。
让我们测试一下比较功能。在终端中运行以下命令:
python3 -c "from stock import Stock; a = Stock('GOOG', 100, 490.1); b = Stock('GOOG', 100, 490.1); c = Stock('AAPL', 200, 123.4); print(f'a == b: {a == b}'); print(f'a == c: {a == c}')"
这段代码创建了三个 Stock
实例:a
、b
和 c
。然后使用 ==
运算符将 a
与 b
以及 a
与 c
进行比较。输出将显示这些比较的结果:
a == b: True
a == c: False
python3 teststock.py
如果一切正常,你应该会看到表明测试通过的输出:
..
----------------------------------------------------------------------
Ran 2 tests in 0.001s
OK
通过仅添加两个简单的方法(__iter__()
和 __eq__()
),我们显著增强了 Structure
类的功能,使其更具 Python 风格且更易于使用。
在编程中,生成器是一个强大的工具,尤其在处理像监控流式数据源这样的实际问题时。在本节中,我们将学习如何将所学的生成器知识应用到这样一个实际场景中。我们将创建一个生成器,用于监控日志文件,并在文件添加新行时将其提供给我们。
在开始创建生成器之前,我们需要设置一个数据源。在这种情况下,我们将使用一个生成股票市场数据的模拟程序。
首先,你需要在 WebIDE 中打开一个新的终端。你将在这个终端中运行启动模拟的命令。
打开终端后,你将运行股票模拟程序。以下是你需要输入的命令:
cd ~/project
python3 stocksim.py
第一个命令 cd ~/project
将当前目录更改为你主目录下的 project
目录。第二个命令 python3 stocksim.py
运行股票模拟程序。这个程序将生成股票市场数据,并将其写入当前目录下名为 stocklog.csv
的文件中。在我们编写监控代码时,让这个程序在后台运行。
现在我们已经设置好了数据源,让我们创建一个程序来监控 stocklog.csv
文件。这个程序将显示所有价格变化为负的情况。
follow.py
的新文件。为此,你需要在终端中使用以下命令将目录更改为 project
目录:cd ~/project
follow.py
文件中。这段代码打开 stocklog.csv
文件,将文件指针移动到文件末尾,然后持续检查是否有新行。如果找到新行,并且它表示价格变化为负,则打印股票名称、价格和变化。## follow.py
import os
import time
f = open('stocklog.csv')
f.seek(0, os.SEEK_END) ## Move file pointer 0 bytes from end of file
while True:
line = f.readline()
if line == '':
time.sleep(0.1) ## Sleep briefly and retry
continue
fields = line.split(',')
name = fields[0].strip('"')
price = float(fields[1])
change = float(fields[4])
if change < 0:
print('%10s %10.2f %10.2f' % (name, price, change))
python3 follow.py
你应该会看到显示价格变化为负的股票的输出。它可能看起来像这样:
AAPL 148.24 -1.76
GOOG 2498.45 -1.55
如果你想停止程序,在终端中按 Ctrl+C
。
虽然之前的代码可以正常工作,但我们可以通过将其转换为生成器函数,使其更具可重用性和模块化。生成器函数是一种特殊类型的函数,它可以暂停和恢复,并一次产生一个值。
follow.py
文件,并将其修改为使用生成器函数。以下是更新后的代码:## follow.py
import os
import time
def follow(filename):
"""
Generator function that yields new lines in a file as they are added.
Similar to the 'tail -f' Unix command.
"""
f = open(filename)
f.seek(0, os.SEEK_END) ## Move to the end of the file
while True:
line = f.readline()
if line == '':
time.sleep(0.1) ## Sleep briefly and retry
continue
yield line
## Example usage - monitor stocks with negative price changes
if __name__ == '__main__':
for line in follow('stocklog.csv'):
fields = line.split(',')
name = fields[0].strip('"')
price = float(fields[1])
change = float(fields[4])
if change < 0:
print('%10s %10.2f %10.2f' % (name, price, change))
follow
函数现在是一个生成器函数。它打开文件,移动到文件末尾,然后持续检查是否有新行。当找到新行时,它会产生该行。
python3 follow.py
输出应该和之前一样。但现在,文件监控逻辑被整齐地封装在 follow
生成器函数中。这意味着我们可以在其他需要监控文件的程序中重用这个函数。
通过将我们的文件读取代码转换为生成器函数,我们使其变得更加灵活和可重用。follow()
函数可以用于任何需要监控文件的程序,而不仅仅是用于股票数据。
例如,你可以使用它来监控服务器日志、应用程序日志或任何随时间更新的其他文件。这展示了生成器是一种以简洁和模块化的方式处理流式数据源的绝佳方法。
在本次实验中,你学习了如何使用生成器在 Python 中自定义迭代。你使用 yield
语句创建了简单的生成器来生成值序列,通过实现 __iter__()
方法为自定义类添加了迭代支持,利用迭代进行了序列转换、解包和比较,并构建了一个用于监控流式数据源的实用生成器。
生成器是 Python 中一个强大的特性,它使你能够用最少的代码创建迭代器。它们在处理大型数据集、处理流式数据、创建数据管道以及实现自定义迭代模式方面特别有用。使用生成器可以让你编写更简洁、更节省内存的代码,清晰地表达你的意图。