Python領域驅動編程實踐-第三章：Flask API及Service層
在上一章中，咱們介紹了，咱們經過Repository的方式來驅動咱們的應用程序，在本章中咱們將討論如何編排業務邏輯，以及業務邏輯和接口代碼的區別。並介紹一些服務層的職責。咱們還要討論一下如何經過服務層與Repository的抽象來讓咱們快速編寫測試。python
咱們將添加一個Flask API與服務層進行交互，它將做爲咱們領域模型的入口點。咱們服務層依賴於AbstractRepository,因此咱們可使用FakeRepository來進行單元測試，而在生產時咱們使用SqlAlchemyRepository運行web
將應用程序鏈接到外部
咱們但願咱們程序快速的在用戶那裏獲得快速反饋，如今咱們有一個領域模型核心部分和分配訂單所需的領域服務函數，還有能夠用於持久化的Repository接口。那麼如今讓咱們把這些東西連接到一塊兒，而後創建一個整潔的架構。下面是咱們的計劃redis
-
使用Flask將咱們的API放置在咱們分配領域服務的前面。而後鏈接數據庫Session與咱們的Repository。以後咱們進行一些簡單的端到端測試。sql
-
構建一個服務層，位於咱們Flask和領域模型之間。構建一些服務層的測試。而後咱們看看怎麼使用咱們的
FakeRepository數據庫 -
爲咱們服務層添加一個參數，能讓與咱們服務層與API層解耦json
第一個端到端測試
對於什麼時端到端測試，什麼是功能測試，什麼是驗收測試，什麼是集成測試，什麼是單元測試。我以爲咱們沒有必要對這種進行論述，也沒有必要爲這個爲陷入爭吵。咱們只將測試分爲快速測試和慢速測試。如今咱們但願編寫一些測試。他們將運行一個真正的客戶端（使用HTTP）與真正的數據庫進行測試。咱們稱之爲端到端測試。下面展現一個例子flask
@pytest.mark.usefixtures('restart_api')
def test_api_returns_allocation(add_stock):
sku, othersku = random_sku(), random_sku('other') #(1)
earlybatch = random_batchref(1)
laterbatch = random_batchref(2)
otherbatch = random_batchref(3)
add_stock([ #(2)
(laterbatch, sku, 100, '2011-01-02'),
(earlybatch, sku, 100, '2011-01-01'),
(otherbatch, othersku, 100, None),
])
data = {'orderid': random_orderid(), 'sku': sku, 'qty': 3}
url = config.get_api_url() #(3)
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 201
assert r.json()['batchref'] == earlybatch
複製代碼
-
random_sku(),random_batchref()等都是小幫助行數，它們使用uuid模塊生成隨機字符。由於咱們如今是針對實際的數據庫運行，因此這是防止各類測試和運行相互干擾的一種方法。api -
add_stock是一個helper fixture(幫助測試的夾具)，它知識隱藏了使用Sql插入數據庫的方法。咱們將再後面的文章中介紹一個更好的辦法。安全 -
config.py是配置文件bash
構建API
如今，咱們用簡單的方式構建
from flask import Flask, jsonify, request
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import config
import model
import orm
import repository
orm.start_mappers()
get_session = sessionmaker(bind=create_engine(config.get_postgres_uri()))
app = Flask(__name__)
@app.route("/allocate", methods=['POST'])
def allocate_endpoint():
session = get_session()
batches = repository.SqlAlchemyRepository(session).list()
line = model.OrderLine(
request.json['orderid'],
request.json['sku'],
request.json['qty'],
)
batchref = model.allocate(line, batches)
return jsonify({'batchref': batchref}), 201
複製代碼
到目前爲止，咱們一切順利。可是咱們尚未commit.咱們實際上並無將數據保存到數據庫中。因此咱們須要第二個測試，來檢若是第一個訂單行已經被分配完了。咱們能不能正確的分配第二個訂單行
@pytest.mark.usefixtures('restart_api')
def test_allocations_are_persisted(add_stock):
sku = random_sku()
batch1, batch2 = random_batchref(1), random_batchref(2)
order1, order2 = random_orderid(1), random_orderid(2)
add_stock([
(batch1, sku, 10, '2011-01-01'),
(batch2, sku, 10, '2011-01-02'),
])
line1 = {'orderid': order1, 'sku': sku, 'qty': 10}
line2 = {'orderid': order2, 'sku': sku, 'qty': 10}
url = config.get_api_url()
# first order uses up all stock in batch 1
r = requests.post(f'{url}/allocate', json=line1)
assert r.status_code == 201
assert r.json()['batchref'] == batch1
# second order should go to batch 2
r = requests.post(f'{url}/allocate', json=line2)
assert r.status_code == 201
assert r.json()['batchref'] == batch2
複製代碼
emmm，不是那麼優雅，由於這逼迫咱們必須進行commit
須要使用真實的數據庫檢查錯誤條件
若是繼續這麼下去，事情會愈來愈糟。
假設咱們想添加一些錯誤處理，若是咱們的領域層發生錯誤，好比產品發生缺貨，該怎麼辦？或者傳入一個根本不存在的產品又怎麼辦？咱們如今的領域甚至不知道，也不該該知道。在調用領域服務以前，咱們應該在數據庫層實現更多的安全性檢查。
因此咱們寫如下兩個端到端測試
@pytest.mark.usefixtures('restart_api')
def test_400_message_for_out_of_stock(add_stock): #(1)
sku, smalL_batch, large_order = random_sku(), random_batchref(), random_orderid()
add_stock([
(smalL_batch, sku, 10, '2011-01-01'),
])
data = {'orderid': large_order, 'sku': sku, 'qty': 20}
url = config.get_api_url()
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 400
assert r.json()['message'] == f'Out of stock for sku {sku}'
@pytest.mark.usefixtures('restart_api')
def test_400_message_for_invalid_sku(): #(2)
unknown_sku, orderid = random_sku(), random_orderid()
data = {'orderid': orderid, 'sku': unknown_sku, 'qty': 20}
url = config.get_api_url()
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 400
assert r.json()['message'] == f'Invalid sku {unknown_sku}'
複製代碼
-
在第一個測試中，咱們試圖分配比庫存更多的產品數量
-
第二中狀況下，SKU根本不存在，對咱們的程序而言，它應該是無效的。
接下來咱們應該在咱們的API處實現它
def is_valid_sku(sku, batches):
return sku in {b.sku for b in batches}
@app.route("/allocate", methods=['POST'])
def allocate_endpoint():
session = get_session()
batches = repository.SqlAlchemyRepository(session).list()
line = model.OrderLine(
request.json['orderid'],
request.json['sku'],
request.json['qty'],
)
if not is_valid_sku(line.sku, batches):
return jsonify({'message': f'Invalid sku {line.sku}'}), 400
try:
batchref = model.allocate(line, batches)
except model.OutOfStock as e:
return jsonify({'message': str(e)}), 400
session.commit()
return jsonify({'batchref': batchref}), 201
複製代碼
如今咱們的應用看起來已經有些重了。咱們的端到端測試數量已經開始失控，很快咱們就會獲得一個倒置的測試金字塔模型，也就是說咱們端到端的測試遠遠大於咱們的單元測試。
引入服務層，並使用FakeRepository進行單元測試
如今咱們看看咱們的API正在作什麼，咱們發現咱們作了太多與咱們API沒有任何關係的東西，好比咱們從數據庫提取東西，針對數據庫狀態驗證咱們的輸入，處理錯誤，而後開心的commit。他們實際上根本用不着進行端到端測試，這也是咱們劃分服務層的意義所在。
還記得咱們第二章構建的FakeRepository嗎？
class FakeRepository(repository.AbstractRepository):
def __init__(self, batches):
self._batches = set(batches)
def add(self, batch):
self._batches.add(batch)
def get(self, reference):
return next(b for b in self._batches if b.reference == reference)
def list(self):
return list(self._batches)
複製代碼
如今就是它大顯身手的時候了，它可讓咱們快速簡單的測試咱們的服務層。
def test_returns_allocation():
line = model.OrderLine("o1", "COMPLICATED-LAMP", 10)
batch = model.Batch("b1", "COMPLICATED-LAMP", 100, eta=None)
repo = FakeRepository([batch]) #(1)
result = services.allocate(line, repo, FakeSession()) #(2)(3)
assert result == "b1"
def test_error_for_invalid_sku():
line = model.OrderLine("o1", "NONEXISTENTSKU", 10)
batch = model.Batch("b1", "AREALSKU", 100, eta=None)
repo = FakeRepository([batch]) #(1)
with pytest.raises(services.InvalidSku, match="Invalid sku NONEXISTENTSKU"):
services.allocate(line, repo, FakeSession()) #(2)(3)
複製代碼
-
FakeRepository保存着咱們的測試將要使用的Batch對象 -
咱們服務模塊(service.py)將定義一個
allocate()服務層函數。它位於咱們API層與咱們領域模型中的allocate()服務函數之間 -
咱們還要一個
Fakesession來僞造數據庫會話。
class FakeSession():
committed = False
def commit(self):
self.committed = True
複製代碼
這個FakeSession是一個臨時解決方法。咱們將在後面介紹另外一種模式來處理。它會更加優雅。如今咱們將咱們的端到端層測試遷移過來
def test_commits():
line = model.OrderLine('o1', 'OMINOUS-MIRROR', 10)
batch = model.Batch('b1', 'OMINOUS-MIRROR', 100, eta=None)
repo = FakeRepository([batch])
session = FakeSession()
services.allocate(line, repo, session)
assert session.committed is True
複製代碼
一個典型的服務函數
咱們將編寫一個相似下面這樣的服務函數
class InvalidSku(Exception):
pass
def is_valid_sku(sku, batches):
return sku in {b.sku for b in batches}
def allocate(line: OrderLine, repo: AbstractRepository, session) -> str:
batches = repo.list() #(1)
if not is_valid_sku(line.sku, batches): #(2)
raise InvalidSku(f'Invalid sku {line.sku}')
batchref = model.allocate(line, batches) #(3)
session.commit() #(4)
return batchref
複製代碼
典型的服務層函數有如下
相似的步驟
-
咱們從Repository中提取一些對象
-
咱們針對當前狀態進行一些檢查和斷言
-
咱們稱之爲服務層
-
若是一切正常，咱們將保存/更新全部已更改的狀態
最後一個步驟有點不太爽，由於咱們的服務層與數據庫進行了耦合。咱們將在後面介紹一種叫unit work的模式來解決它。讓它依賴於抽象
關於咱們的服務層函數還有一點須要注意
def allocate(line: OrderLine, repo: AbstractRepository, session) -> str:
複製代碼
它顯式的依賴於一個Repository，而且用一個type hint來提示咱們依賴的是一個AbstractRepository。因此當咱們測試的時候，不管給他一個FakeRepository仍是給它一個SqlalchemyRepository時，他均可以工做。
若是你還記得咱們的DIP(依賴倒置原則)，如今就能夠看到咱們所說的咱們應該依賴於抽象的意思 咱們高級模塊(Service層)依賴於Repository的抽象。咱們一些別的Repository的實現細節也應該依賴於相同的抽象。好比咱們加一個什麼Mongodb/Redis/CSV等等。
如今咱們的應用看起來就乾淨多了
@app.route("/allocate", methods=['POST'])
def allocate_endpoint():
session = get_session() #(1)
repo = repository.SqlAlchemyRepository(session) #(1)
line = model.OrderLine(
request.json['orderid'], #(2)
request.json['sku'], #(2)
request.json['qty'], #(2)
)
try:
batchref = services.allocate(line, repo, session) #(2)
except (model.OutOfStock, services.InvalidSku) as e:
return jsonify({'message': str(e)}), 400 (3)
return jsonify({'batchref': batchref}), 201 (3)
複製代碼
-
咱們實例化一個數據庫會話和一些
Repository對象 -
咱們從web請求中提取了用戶的參數，而且傳遞給領域服務
-
咱們返回適當的狀態代碼和一些JSON響應
Flask的職責僅僅包含了標準web的東西：對Web Session進行管理、從請求中解析參數，返回狀態碼和JSON。咱們全部的業務流程邏輯都位於咱們的Service層。
最後呢，咱們把咱們多餘的端到端測試幹掉。只包含兩個，一個正確的一個錯誤的
@pytest.mark.usefixtures('restart_api')
def test_happy_path_returns_201_and_allocated_batch(add_stock):
sku, othersku = random_sku(), random_sku('other')
earlybatch = random_batchref(1)
laterbatch = random_batchref(2)
otherbatch = random_batchref(3)
add_stock([
(laterbatch, sku, 100, '2011-01-02'),
(earlybatch, sku, 100, '2011-01-01'),
(otherbatch, othersku, 100, None),
])
data = {'orderid': random_orderid(), 'sku': sku, 'qty': 3}
url = config.get_api_url()
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 201
assert r.json()['batchref'] == earlybatch
@pytest.mark.usefixtures('restart_api')
def test_unhappy_path_returns_400_and_error_message():
unknown_sku, orderid = random_sku(), random_orderid()
data = {'orderid': orderid, 'sku': unknown_sku, 'qty': 20}
url = config.get_api_url()
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 400
assert r.json()['message'] == f'Invalid sku {unknown_sku}'
複製代碼
這樣咱們成功的將咱們的測試分紅了兩類：一類是關於web的測試，咱們實現端到端測試。另外一類是關於咱們領域內容的單元測試。
爲何全部都叫作Service
如今，我估計不少人已經有些撓頭，試圖弄明白，領域服務與服務層有啥不一樣。在本章中，咱們使用了兩種稱爲Service的東西。第一個是應用程序服務(服務層)。它的職責是處理來自外部的請求並安排操做。服務層通常作下面簡單的步驟來驅動應用程序
-
從數據庫中獲取一些數據
-
更新領域模型
-
持久化更改
對於系統中的每一個操做，這是一個很是枯燥無聊的活，可是將其與業務邏輯分離有助於保持咱們應用程序乾淨整潔。
第二類服務是領域服務。這是個服務屬於領域服務，它主要職責是操做實體與值對象之間的邏輯，好比：你作了一個購物者程序，您可能會選擇將優惠券構建爲一個領域服務。計算優惠與更新購物車是不一樣的工做。也是模型組成的重要部分。可是爲這項工做設置一個實體其實也不太合適。取而代之的是一個CouponCalculator類或者一個calculator_conpon的函數就完事了。
組織咱們的應用程序
如今隨着咱們的應用程序愈來愈大，咱們須要整理咱們的目錄結構。下面介紹一下項目佈局參考
.
├── config.py
├── domain #(1)
│ ├── __init__.py
│ └── model.py
├── service_layer #(2)
│ ├── __init__.py
│ └── services.py
├── adapters #(3)
│ ├── __init__.py
│ ├── orm.py
│ └── repository.py
├── entrypoints (4)
│ ├── __init__.py
│ └── flask_app.py
└── tests
├── __init__.py
├── conftest.py
├── unit
│ ├── test_allocate.py
│ ├── test_batches.py
│ └── test_services.py
├── integration
│ ├── test_orm.py
│ └── test_repository.py
└── e2e
└── test_api.py
複製代碼
-
爲咱們的領域模型設置一個文件夾
-
將咱們應用服務設立一個文件夾，這裏咱們也能夠添加一些咱們服務層的異常錯誤
-
adapters是咱們圍繞外部IO創建的抽象，好比redis_client等等
-
Entrypoints是咱們API入口的地方，經典的MVC架構中，咱們也能夠稱之爲View層。
總結
增長應用服務層帶給咱們不少好處
-
咱們的API變得很是薄，且容易編寫：他們惟一的職責就是處理Web相關的事情
-
咱們爲領域定義了一個清晰的接口。一組用例或入口點，任何適配器均可以使用這些入口點，不管是cli api 仍是什麼東西
-
使用咱們服務層咱們能夠快速的編寫測試，咱們也能夠很是大膽的重構咱們的領域模型。咱們就能夠嘗試新的設計，而不會由於這樣重寫大量的測試。
-
咱們的測試看起來也不錯，咱們大部分的測試是針對的服務的單元測試，只有極少數的端到端測試和集成測試。單元測試不鏈接真實的數據庫，因此速度很是快。這也對咱們的CI/CD有極大的好處。
DIP在行動
服務層的抽象依賴表達了咱們服務層與領域服務層的依賴：領域模型和AbstractRepository。當咱們運行測試時，測試提供了一個抽象依賴的實現也就是咱們的FakeRepository,當咱們運行實際的應用程序時，咱們又把他替換成了真實的在咱們的例子中是SqlalchemyRepository
+-----------------------------+
| Service Layer |
+-----------------------------+
| |
| | depends on abstraction
V V
+------------------+ +--------------------+
| Domain Model | | AbstractRepository |
| | | (Port) |
+------------------+ +--------------------+
複製代碼
+-----------------------------+
| Tests |-------------\
+-----------------------------+ |
| |
V |
+-----------------------------+ |
| Service Layer | provides |
+-----------------------------+ |
| | |
V V |
+------------------+ +--------------------+ |
| Domain Model | | AbstractRepository | |
+------------------+ +--------------------+ |
^ |
implements | |
| |
+----------------------+ |
| FakeRepository |<--/
| (in–memory) |
+----------------------+
複製代碼
+--------------------------------+
| Flask API (Presentation Layer) |-----------\
+--------------------------------+ |
| |
V |
+-----------------------------+ |
| Service Layer | |
+-----------------------------+ |
| | |
V V |
+------------------+ +--------------------+ |
| Domain Model | | AbstractRepository | |
+------------------+ +--------------------+ |
^ ^ |
| | |
gets | +----------------------+ |
model | | SqlAlchemyRepository |<--/
definitions| +----------------------+
from | | uses
| V
+-----------------------+
| ORM |
| (another abstraction) |
+-----------------------+
|
| talks to
V
+------------------------+
| Database |
+------------------------+
複製代碼
如今咱們暫停討論服務層了，該到咱們權衡利弊的時候了。
| 優勢 | 缺點 |
|---|---|
| 咱們只有一個地方能夠找到應用程序的業務邏輯 | 若是是一個MVC架構的應用程序，那麼你的controllers/view就是捕獲全部業務邏輯的惟一位置 |
| 咱們已經把領域邏輯放到了API以後，這樣咱們就能夠持續重構了 | 這是另外一個抽象層 |
| 咱們乾淨的將Web相關的東西與業務邏輯的東西區分開了 | 將太多的業務邏輯放入服務層會致使一個貧血領域的反模式。最好時在業務邏輯已經滲透進控制器的時候引入這一層 |
結合了Repository和FakeRepository後，咱們能夠比領域層更高的層次上編寫測試，咱們可單元測試不少工做流，而不是集成測試這種 |
你原本能夠從富模型中得到更多益處，只須要將邏輯從控制器推到模型層就好了，而不是再添加一層，這樣增大了局部複雜度。即（胖模型，瘦控制器） |
固然咱們仍有一些很差的地方須要收拾
-
服務層仍然與領域層緊密耦合，由於API是用OrderLine這個領域模型進行表示的。在下章中咱們將解決這個問題，並討論服務層如何提升TDD的效率。
-
服務層與數據庫session緊密耦合。在咱們講解unit of work模式中，咱們會介紹另外一個存儲庫和服務層協做的新模式。
- 1. Python領域驅動編程實踐-第一章：領域建模
- 2. 【DDD】領域驅動設計實踐
- 3. 領域驅動設計和實踐
- 4. DDD領域驅動實踐記錄
- 5. EntityFramework之領域驅動設計實踐
- 6. 領域驅動戰略設計實踐
- 7. 領域驅動設計實踐
- 8. 實踐DDD領域驅動設計
- 9. DDD 領域驅動設計 實踐03
- 10. 領域驅動設計最佳實踐
- 更多相關文章...
- • Web Service 實例 - Web Services 教程
- • Thymeleaf項目實踐 - Thymeleaf 教程
- • 三篇文章瞭解 TiDB 技術內幕 —— 說計算
- • 三篇文章瞭解 TiDB 技術內幕——說存儲
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. Python領域驅動編程實踐-第一章：領域建模
- 2. 【DDD】領域驅動設計實踐
- 3. 領域驅動設計和實踐
- 4. DDD領域驅動實踐記錄
- 5. EntityFramework之領域驅動設計實踐
- 6. 領域驅動戰略設計實踐
- 7. 領域驅動設計實踐
- 8. 實踐DDD領域驅動設計
- 9. DDD 領域驅動設計 實踐03
- 10. 領域驅動設計最佳實踐