用 python flask 開發 SOAP 界面
需求說明
財務部想取得商家庫存自動產生報表。然而財務部的 SAP 系統是走 SOAP ，現有的庫存 API 則是 RESTful JSON API。需要一個中間件聽取 SAP 的請求，造訪庫存 API 之後再翻譯成 SOAP 傳給 SAP 。
蛤？ SOAP，如果你是位幸運的開發者，祝福你不會遇到這個協定。如果很不幸遇到了，希望這篇文章能幫上小小的忙。
由於這個項目大部分依賴企業內部現有資訊架構，我們遇到的挑戰也會與其相關。最大的兩個挑戰是：
- 財務部 SAP 系統使用 SOAP，如何串接這個協定？
- 公司不是使用像 AWS 之類的雲端主機服務，我們必須佈署在有其他系統運行的主機上。如何在不影響其他系統運作的情況下，讓我們開發的應用程式順利運作呢？
粗略的說，兩個星期半的時間，第一個星期我們研究如何做 SOAP 的接口，第二個星期到最後一天同伴研究事件日誌，我研究佈署。當然，還有很多時間花在溝通、開會、寫文件。
中間件用了 python flask 實作。用 Flask-Spyne 套件做出 SOAP 接口，用 unittest 、Flask-Testing 的 LiveServerTestCase 做測試，用 suds 和 SOAP UI 模擬使用 SOAP API 狀況。
關於佈署的細節，可見下篇文章「用 Docker + pyinstaller 建置執行檔」。
SOAP接口的細節
在百度 Python + SOAP 之後，應該能夠很快的找到 spyne 這個解方（或 flask-spyne），官方範例大致如下：
app = Flask(__name__)
spyne = Spyne(app)class SomeSoapService(spyne.Service):
__service_url_path__ = ‘/soap/someservice’
__in_protocol__ = Soap11(validator=’lxml’)
__out_protocol__ = Soap11(polymorphic=False)@spyne.srpc(Unicode, Integer, _returns=Iterable(Unicode))
def echo(str, cnt):
for i in range(cnt):
yield str
這讓 flask app 能夠和 SOAP 對接，只要造訪 /soap/someservice 就會出現 SOAP 要的 wsdl 檔。wsdl 是個 xml 檔案，他把所有這個服務能呼叫的函式像目錄一樣列出來，而且函式參數是數字、文字、還是其他型別都要指定清楚。
但根據官網範例，你大概只能寫出回傳 cnt 次文字的函數。大多時候的需求是要回傳巢狀結果，如圖：
解答是 polymorphic_array，要用到 ComplexModel 這個類，範例躲在 spyne 的原始碼裡不太好找，請點參考連結進去看。
測試
flask 是很好測試的 framework，這次專案有做到幾個測試：
- 把使用的套件(Spyne、suds、庫存 API)官網範例寫成測試，讓維護人員可以很快上手套件用法
- 幾個功能作到測試驅動開發
- 用 flask-testing 的 LiveServerTestCase，可以做出類似 phantomJS 的測試效果
- flask 原生的測試 main.app.test_client()
- 有因為寫測試抓到庫存 API 壞掉的時候 （API 開發者剛好在維護）
希望下次有機會能用用 pytest
LiveServerTestCase 結合 suds client 使用範例
測試以 class 方式組織。在開始測試的時候，會先執行 create_app 這個函式，接著依次測試每個 test_ 開頭的方法。可用 setUp 與 tearDown 定義每次執行 test_ 開頭方法的前後需要重複做的事情。
這裡利用了 setUp 方法來偷偷創建 suds client 的物件。
class SapMiddlewareTest(LiveServerTestCase): def create_app(self):
# 這個函式必須定義
app = main.spyne.app
app.config[‘TESTING’] = True
app.config[‘LIVESERVER_PORT’] = 5566
return app def setUp(self):
# 這個函式會在下面每個 test 開頭的方法之前重新執行一次
main.init_db()
url = urljoin(self.get_server_url(), “/xmlapi?wsdl”)
self.client = SudsClient(url=url, cache=None) def test_server_is_up_and_running(self):
response = self.client.service.echo()
self.assertEqual(str(response), “Service up and running”)
參考資料
財務部中間件：
SOAP 細節：
測試：
下一篇：
