将yield from视为提供了一个调用者和子生成器之间的透明的双向通道。包括从子生成器获取数据以及向子生成器传送数据。
def reader():
# 模拟从文件读取数据的生成器
for i in range(4):
yield '<< %s' % i
def reader_wrapper(g):
# 循环迭代从reader产生的数据
for v in g:
yield v
wrap = reader_wrapper(reader())
for i in wrap:
print(i)
# 结果：
<< 0
<< 1
<< 2
<< 3
我们可以用yield from语句替代reader_wrapper(g)函数中的循环，如下
def reader():
# 模拟从文件读取数据的生成器
for i in range(4):
yield '<< %s' % i
def reader_wrapper(g):
yield from g # 执行 g生成器
wr = reader_wrapper(reader())
for i in wr:
print(i)
利用yield from语句向生成器（协程）传送数据
首先创建一个生成器writer，接收传送给它的数据，并写进套接字，文件等
def writer():
# 读取send传进的数据，并模拟写进套接字或文件
while True:
w = (yield) # w接收send传进的数据
print('>> ', w)
现在的问题是，包装器函数如何传送数据给writer函数，使得传递给包装器的数据都能够显示地传递给writer函数？即我们期待得到如下结果：
def writer_wrapper(coro):
# TBD
pass
w = writer()
wrap = writer_wrapper(w)
wrap.send(None) # 生成器准备好接收数据
for i in range(4):
wrap.send(i)
# 期望结果：
>> 0
>> 1
>> 2
>> 3
很显然，包装区需要接收数据并显示传递给生成器，并且需要处理for循环耗尽是生成器产生的StopIteration异常，显然包装器只用for循环已经不能满足需求，满足情况的一般版本如下：
def writer():
# 读取send传进的数据，并模拟写进套接字或文件
while True:
w = (yield) # w接收send传进的数据
print('>> ', w)
def writer_wrapper(coro1):
coro1.send(None) # 生成器准备好接收数据 初始化
while True:
try:
x = (yield) # x接收send传进的数据
coro1.send(x) # 然后将x在send给writer子生成器
except StopIteration as e: # 处理子生成器返回的异常
print(e.value)
break
"""
g = writer_wrapper
g.send(value) # 如何让value传值 传给 writer()装饰器
"""
w = writer()
g = writer_wrapper(w)
g.send(None) # x 会接收到
for i in range(6):
g.send(i)
结果
>> 0
>> 1
>> 2
>> 3
>> 4
>> 5
包装器也是个生成器，上面所有复杂的写法也可以用yield from替换：
# 一下子少了好多代码，是不是见证了奇迹！
def writer_wrapper(coro2):
yield from coro2
def writer():
# 读取send传进的数据，并模拟写进套接字或文件
while True:
w = (yield) # w接收send传进的数据
print('>> ', w)
def writer_wrapper(coro1):
yield from coro1
w = writer()
g = writer_wrapper(w)
g.send(None) # w 会接收到 这一步初始化 必须做，不然会报错
# TypeError: can't send non-None value to a just-started generator
for i in range(6):
g.send(i)
结果
>> 0
>> 1
>> 2
>> 3
>> 4
>> 5
总结
关于
yield的用法，今天先给大家讲到这里，关于yield的其他用法，明天还有一篇文章来进行讲解，剧透一下，是关于yield异常处理的。
实际例子：
获取json数据的key的位置
import json
from typing import List
class JsonPathFinder:
def __init__(self, json_str):
self.data = json.loads(json_str)
def iter_node(self, data, road_step, target):
if isinstance(data, dict):
key_value_iter = (x for x in data.items())
elif isinstance(data, list):
key_value_iter = (x for x in enumerate(data))
else:
return
for key, value in key_value_iter:
current_path = road_step.copy()
current_path.append(key)
if key == target:
yield current_path
if isinstance(value, (dict, list)):
yield from self.iter_node(value, current_path, target) # 在函数里面调用生成器用yield from
def find_one(self, key: str) -> list:
path_iter = self.iter_node(self.data, [], key)
for path in path_iter:
return path
return []
def find_all(self, key) -> List[list]:
path_iter = self.iter_node(self.data, [], key)
return list(path_iter)
if __name__ == '__main__':
file_data = {
"a1key":
{"bkey":
{"ckey":
{"dkey":
{"loadid":1}
}}},
"a2key":
{"bkey":
{"ckey":
{"dkey": {
"loadid": 1}
}}},
"a3key": {"bkey": [1, 5, 3, ["loadid"]]},
"a4key": {"bkey": [1, 5, 3, [{"loadid": 555}]]}
}
file_data = json.dumps(file_data, ensure_ascii=False,indent=2)
finder = JsonPathFinder(file_data)
path_list = finder.find_all('loadid')
print(f"长度为{len(path_list)}")
data = finder.data
for path in path_list:
print(path)
# from jsonpath import jsonpath
# datas = json.load(open("sample.json", mode="r"))
# print(jsonpath(datas, "$..full_text"))
# print(len(jsonpath(datas, "$..full_text")))
#
# for i in jsonpath(datas, "$..full_text"):
# if i.startswith("Want to get a cool Python Gift"):
# print(i)
# print(type(i))
# pass
网友评论