Python 的 yield from 語法簡化了生成器的實作，並強化了與子生成器的互動。它不僅能串接多個生成器，還能有效地捕捉子生成器的傳回值，並實作雙向資料傳遞和例外處理。隨著 Python 的發展，async 和 await 語法出現，提供了更清晰的非同步程式設計模型。然而，理解 yield from 的運作機制仍然有助於掌握 Python 非同步程式設計的底層邏輯。此外，文章也介紹了非同步迭代器和產生器的使用，以及如何利用 anext 函式推進非同步迭代器。最後，文章也強調了單元測試的重要性，並示範瞭如何使用 unittest 和 pytest 框架編寫單元測試，確保程式碼的品質和可維護性。
yield from 語法解析與應用
在 Python 中，yield from 語法是一種強大的工具，用於簡化生成器（generator）的實作並提升其功能。本文將探討 yield from 的基本概念、捕捉子生成器傳回值、以及如何與子生成器進行資料傳遞。
基本概念與簡化生成器實作
yield from 語法主要用於簡化巢狀生成器的實作。一個典型的例子是實作類別似於標準函式庫中的 itertools.chain() 函式。該函式允許傳入多個可迭代物件，並將它們合併成一個單一的迭代流。
原始實作
def chain(*iterables):
for it in iterables:
for value in it:
yield value
使用 yield from 簡化
def chain(*iterables):
for it in iterables:
yield from it
兩種實作方式具有相同的行為，能夠正確地將多個可迭代物件合併成一個迭代流。
內容解密：
yield from it直接將it中的值產生出來，無需巢狀迴圈。- 這種語法簡化了程式碼，使其更易讀且更具 Python 風格。
捕捉子生成器的傳回值
yield from 不僅能簡化生成器的實作，還能捕捉子生成器的傳回值。以下是一個範例，展示瞭如何使用 yield from 捕捉子生成器的傳回值。
範例程式碼
def sequence(name, start, end):
logger.info("%s started at %i", name, start)
yield from range(start, end)
logger.info("%s finished at %i", name, end)
return end
def main():
step1 = yield from sequence("first", 0, 5)
step2 = yield from sequence("second", step1, 10)
return step1 + step2
內容解密：
sequence生成器函式使用yield from range(start, end)產生一系列數值，並在結束時傳回end值。- 在
main函式中，yield from sequence(...)不僅產生sequence中的值，還捕捉其傳回值並指定給step1和step2。 - 最終，
main傳回step1和step2的總和。
與子生成器進行資料傳遞
除了捕捉傳回值，yield from 還允許向子生成器傳送資料或丟擲異常。以下範例展示瞭如何修改 sequence 生成器以接收資料和處理異常。
修改後的 sequence 生成器
def sequence(name, start, end):
value = start
logger.info("%s started at %i", name, value)
while value < end:
try:
received = yield value
logger.info("%s received %r", name, received)
value += 1
except CustomException as e:
logger.info("%s is handling %s", name, e)
received = yield "OK"
return end
內容解密：
sequence生成器現在能夠接收透過send()傳送的值，並在日誌中記錄。- 當透過
throw()丟擲CustomException時，生成器能夠捕捉並處理該異常。 - 在處理異常後，生成器產生一個
"OK"值表示處理成功。
非同步程式設計的演進與 Python 中的實作
在 Python 中，非同步程式設計的概念隨著時間不斷演進。早期，開發者利用生成器（generators）和 yield from 語法來實作非同步操作。這些技術讓開發者能夠建立並管理多個協程（coroutines），並在適當的時候切換它們，以達到非同步處理的效果。
早期實作：使用生成器和 yield from
在 Python 3.5 之前，開發者使用生成器來實作協程，並利用 yield from 語法來傳遞值或異常給子協程。這種方法雖然有效，但存在一些問題。由於生成器和協程在技術上是相同的，這導致開發者可能會將這兩種不同的物件混淆，進而在執行時才發現錯誤。
def main():
step1 = yield from first_coroutine()
step2 = yield from second_coroutine(step1)
return step2
#### 內容解密：
此段程式碼展示瞭如何使用 `yield from` 將控制流交給子協程，並接收它們的傳回值。首先，`main` 協程呼叫 `first_coroutine` 並等待其完成，然後將結果傳遞給 `second_coroutine`。這裡的關鍵在於 `yield from` 語法允許 `main` 協程暫停執行，直到子協程完成。
### 新語法：`async def` 和 `await`
為瞭解決上述問題，Python 3.5 引入了新的語法：`async def` 用於定義協程，`await` 用於取代 `yield from`。這種新語法提高了程式碼的可讀性和安全性，因為它明確區分了協程和生成器。
```python
async def main():
step1 = await first_coroutine()
step2 = await second_coroutine(step1)
return step2
#### 內容解密：
這段程式碼展示瞭如何使用 `async def` 定義協程，並利用 `await` 語法等待子協程的完成。與 `yield from` 類別似，`await` 會暫停當前協程的執行，直到被等待的協程完成。這種語法更加直觀，也更不容易出錯。
### 事件迴圈與非同步程式設計
Python 中的非同步程式設計依賴於事件迴圈（event loop）來管理協程的執行。事件迴圈負責排程協程，並在適當的時候切換它們。開發者可以利用 `asyncio` 函式庫來建立和管理事件迴圈。
```python
import asyncio
async def main():
# 非同步操作
await asyncio.sleep(1)
print("非同步操作完成")
asyncio.run(main())
#### 內容解密：
此範例展示瞭如何使用 `asyncio` 函式庫來執行非同步操作。`asyncio.run(main())` 用於啟動事件迴圈並執行 `main` 協程。在 `main` 協程中，`await asyncio.sleep(1)` 暫停執行 1 秒，模擬非同步操作。
## 深入理解Python中的非同步程式設計
Python中的`async def`與`await`語法為開發者提供了一種與事件迴圈（event loop）互動的API。預設情況下，這個事件迴圈通常是標準函式庫中的`asyncio`，但任何符合API規範的事件迴圈系統都可以使用。這意味著開發者可以使用像`uvloop`和`trio`這樣的函式庫，並且程式碼仍然能夠正常運作。
### 神奇的非同步方法
在之前的章節中，我們已經瞭解到利用Python的神奇方法（magic methods）可以使我們建立的抽象概念自然地融入語言的語法中，從而實作更好、更簡潔的程式碼。但是，當我們需要在這些方法中呼叫協程（coroutine）時，情況就變得不同了。如果一個函式中需要使用`await`，那麼這個函式本身就必須被定義為一個協程（使用`async def`），否則就會出現語法錯誤。
為瞭解決這個問題，Python引入了新的語法和新的神奇方法。這些新的神奇方法與原有的語法是類別似的。
#### 非同步上下文管理器
非同步上下文管理器的概念很簡單：如果我們需要使用一個上下文管理器，但是需要對它呼叫一個協程，那麼我們就不能使用普通的`__enter__`和`__exit__`方法，因為它們是被定義為普通的函式。因此，我們需要使用新的`__aenter__`和`__aexit__`協程方法。並且，我們需要使用`async with`語法來呼叫它。
```python
@contextlib.asynccontextmanager
async def db_management():
try:
await stop_database()
yield
finally:
await start_database()
async def run_db_backup():
async with db_management(), metrics_logger():
print("Performing DB backup...")
非同步迭代
與普通的迭代器物件類別似，我們也可以建立非同步迭代器。非同步迭代器支援使用Python內建的async for迴圈進行迭代。如果我們想要建立一個迭代器來抽象化從外部來源讀取資料的方式，但是提取資料的部分是一個協程，那麼我們就需要使用__anext__協程。
class AsyncIterator:
def __init__(self, data):
self.data = data
async def __aiter__(self):
return self
async def __anext__(self):
if not self.data:
raise StopAsyncIteration
item = self.data.pop(0)
# 假設這裡有一個協程來處理item
await process_item(item)
return item
async def main():
data = [1, 2, 3]
async for item in AsyncIterator(data):
print(item)
程式碼解析
在上述範例中，我們定義了一個非同步迭代器AsyncIterator，它實作了__aiter__和__anext__方法。在__anext__方法中，我們模擬了一個協程來處理每個專案。當資料列表為空時，我們引發了StopAsyncIteration異常來停止迭代。
在main函式中，我們使用async for迴圈來迭代AsyncIterator例項，並列印出每個專案。
非同步迭代器與產生器在Python中的應用
在Python的非同步程式設計中，迭代器和產生器是兩個非常重要的概念。它們讓我們能夠以非同步的方式處理資料流，使得程式設計更加靈活和高效。
非同步迭代器
非同步迭代器是一種特殊的物件，它允許我們在非同步程式中使用async for迴圈來迭代資料。下面是一個簡單的非同步迭代器範例：
import asyncio
import random
async def coroutine():
await asyncio.sleep(0.1)
return random.randint(1, 10000)
class RecordStreamer:
def __init__(self, max_rows=100) -> None:
self._current_row = 0
self._max_rows = max_rows
def __aiter__(self):
return self
async def __anext__(self):
if self._current_row < self._max_rows:
row = (self._current_row, await coroutine())
self._current_row += 1
return row
raise StopAsyncIteration
內容解密：
__aiter__方法傳回迭代器物件本身，這是實作非同步迭代器的必要步驟。__anext__方法是核心，它負責傳回下一個值。如果沒有更多的值，則丟出StopAsyncIteration例外。- 在
__anext__中，我們使用await coroutine()來非同步地取得資料。
使用非同步迭代器：
async for row in RecordStreamer(10):
print(row)
推進非同步迭代器
由於非同步迭代器不支援 next() 函式，我們需要使用不同的方式來推進它。
NOT_SET = object()
async def anext(async_generator_expression, default=NOT_SET):
try:
return await async_generator_expression.__anext__()
except StopAsyncIteration:
if default is NOT_SET:
raise
return default
內容解密：
anext函式模擬了next()的行為，但適用於非同步迭代器。- 它嘗試取得下一個值，如果沒有更多的值，則根據是否有預設值來決定是否丟出例外。
使用範例：
$ python -m asyncio
>>> streamer = RecordStreamer(10)
>>> await anext(streamer)
(0, 5017)
>>> await anext(streamer)
(1, 5257)
非同步產生器
Python 3.6 之後引入了非同步產生器，讓我們能夠以更簡潔的方式實作非同步迭代。
async def record_streamer(max_rows):
current_row = 0
while current_row < max_rows:
row = (current_row, await coroutine())
current_row += 1
yield row
內容解密：
- 使用
async def定義非同步產生器。 - 在迴圈內，使用
yield來產生值。 - 非同步產生器比非同步迭代器更簡潔，易於理解和維護。
單元測試與重構：軟體開發的根本
單元測試是軟體開發中不可或缺的一環，它們對於確保程式碼的品質和可維護性至關重要。在本章中，我們將探討單元測試的重要性、運作原理，以及如何利用單元測試來提升程式碼的品質。
設計原則與單元測試
在探討單元測試的實踐之前，我們需要先了解其背後的設計原則。單元測試不僅是驗證程式碼正確性的工具，更是軟體工程設計原則的體現。
單元測試的定義
單元測試是負責驗證程式碼其他部分的程式碼。它們通常被視為軟體開發的核心組成部分，而非次要元素。一個單元測試會匯入包含業務邏輯的程式碼，並執行其邏輯，以斷言多種場景，確保特定的條件得到滿足。
單元測試的特性
單元測試必須具備以下特性：
- 隔離性：單元測試應完全獨立於任何外部代理，只關注業務邏輯。這意味著它們不應連線到資料函式庫、不應執行HTTP請求等。隔離性還意味著測試之間是獨立的，它們可以按任意順序執行，而不依賴於任何先前的狀態。
實踐單元測試
在瞭解了單元測試的概念和特性之後，接下來我們將探討如何在實踐中應用單元測試。
使用unittest模組
Python的標準函式庫中提供了unittest模組，用於編寫和執行單元測試。下面是一個簡單的例子：
import unittest
def add(a, b):
return a + b
class TestAddFunction(unittest.TestCase):
def test_add(self):
self.assertEqual(add(1, 2), 3)
self.assertEqual(add(-1, 1), 0)
self.assertEqual(add(-1, -1), -2)
if __name__ == '__main__':
unittest.main()
內容解密：
在這個例子中，我們定義了一個簡單的add函式，並建立了一個TestAddFunction類別來測試它。test_add方法包含了多個斷言，用於驗證add函式在不同輸入下的行為是否正確。
使用pytest框架
除了unittest模組外，pytest是一個流行的外部套件，用於編寫和執行單元測試。它提供了更多的功能和彈性，使得編寫測試更加方便。
安裝pytest
pip install pytest
編寫pytest測試
def add(a, b):
return a + b
def test_add():
assert add(1, 2) == 3
assert add(-1, 1) == 0
assert add(-1, -1) == -2
內容解密：
在這個例子中，我們使用pytest框架編寫了一個測試函式test_add。pytest自動發現並執行以test_開頭的函式，簡化了測試的編寫和執行過程。
