AWS Lambdaからsuds-py3を使ってSOAP API連携をしてみた

RESTにとって変わられがちですが、ちょっと触る機会があったので、
表題の件を簡単かつざっくり解説していきます。

なお連携対象はOracle B2C ServiceのSOAP APIの想定です。

##環境
Python3.8
AWS Lambda プロキシ統合あり

ライブラリ
suds-py3 1.4.4.1

##準備

まずはライブラリをインストール

$ pip3 install suds-py3

##実装

内容は連携するAPIに準じて読み替えてください。
インポート関係はこんな感じです。

import os
import json
from suds.client import Client
from suds.wsse import *
from suds.sudsobject import asdict

WSDL = os.environ['WSDL'] # 指定のWSDLのURLを記載
APP_ID = os.environ['APP_ID']
USERNAME = os.environ['USERNAME']
PASSWORD = os.environ['PASSWORD']

###URLからクライアントの作成
WSDLのURLでサービスのクライアントを作成します

client = Client(WSDL)

###WS-Securityの設定
続いてWS-Securityを設定します。
WS-Securityについてはこちらを参照

security = Security()
token = UsernameToken(USERNAME, PASSWORD)
security.tokens.append(token)
client.set_options(wsse=security)

###ヘッダーの作成
ヘッダーも作成していきます。

client_info_header = client.factory.create('rnm_v1:ClientInfoHeader')
client_info_header.AppID = APP_ID
client.set_options(soapheaders=client_info_header)

※カスタムSOAPヘッダーで作成することもできるので試してみましたが、
たまにXMLが壊れることがあったので、きちんとfactory.createでWSDLで定義されたオブジェクトとタイプのインスタンスを作成することをオススメします。

このようにXML文字列としてヘッダーに渡そうとすると、
エスケープされるとのことなのでやめましょう。

client = client(url)
ssn = '<ssn:SessionID>123</ssn:SessionID>'
client.set_options(soapheaders=ssn)
result = client.service.addPerson(person)

###サービス実行

interactionIDを生成するためにStartInteractionのサービスを実行します。

interaction_id = client.service.StartInteraction(app_identifier, user_ip_address)

せっかくなので、interactionIDを使って、
パラメータを取得してみます。

response = client.service.SearchContent(interaction_id, search_terms, None, None, None, limit, None, None, None, start)

print(type(response))

結果

<class 'suds.sudsobject.SearchResponse'>

流石にこのままだと使いづらいので、

def recursive_dict(d):
    """
    取得したレスポンスを辞書型に整形
    """
    out = {}
    for k, v in asdict(d).items():
        if hasattr(v, '__keylist__'):
            out[k] = recursive_dict(v)
        elif isinstance(v, list):
            out[k] = []
            for item in v:
                if hasattr(item, '__keylist__'):
                    out[k].append(recursive_dict(item))
                else:
                    out[k].append(item)
        else:
            out[k] = v
    return out

こんな感じで使いやすい形に整形して返してあげましょう。

##参考
https://github.com/cackharot/suds-py3