注文データ（CSV）をデータベースへ登録するフローを題材に、エラーの種類ごとに「止める」「通知して続ける」を使い分けるエラー処理を作成してみました。
これまでの記事で紹介したException／ExceptionReturn、メール送信、Slack通知を1本の業務フローにつないでいます。
※動作確認環境：ASTERIA Warp 2608、Windows 11
エラー処理を決めずに作ると何が起きるか
エラー処理を設定していないフローは、どこか1か所でエラーが起きるとその場で異常終了します。
CSVの1行に誤りがあるだけで残りのデータが登録されなかったり、逆に、止めるべき異常なデータが登録されてしまったりします。
そこで、作り始める前に、エラーの種類ごとに「止めるか、続けるか」と「誰にどう知らせるか」を決めておきます。
今回のフローでは、次のように決めました。
・入力ファイルが無い：止める／メールで通知（パターン2）
・顧客コードの形式が誤っている行：その行だけエラーファイルに退避し、続ける（パターン1）
・金額が上限（1,000,000円）を超える行：業務ルール違反として止める／メールで通知（パターン4）
・DB登録に失敗した行（注文IDの重複など）：その行だけスキップし、Slackで通知して続ける（パターン3）
※「止める」エラーはすぐに人が対応すべきものなのでメール、「続ける」エラーは後で確認すればよいものなのでSlack、という分け方にしています。
********************************************
テストデータとテーブル
入力ファイルは、DEMOユーザーのホームフォルダーに置いた次のCSVです（データはダミーです）。
注文ID,顧客コード,商品コード,数量,金額
O0001,C001,P-100,2,3000
O0002,C002,P-200,1,15000
O0003,X99,P-300,1,2000
O0002,C004,P-500,1,8000
O0006,C005,P-600,2,4000・O0003：顧客コードの形式エラー（パターン1の確認用）
・4行目のO0002：注文IDの重複によるDB登録エラー（パターン3の確認用）
・パターン4の確認には、最後に「O0007,C006,P-700,3,1500000」を加えたファイルを使います。
登録先は、インターナルDB（asteria_internal）に作成した次のテーブルです。
CREATE TABLE d1_orders (
order_id VARCHAR(10) PRIMARY KEY,
customer_code VARCHAR(10),
item_code VARCHAR(20),
qty INTEGER,
amount INTEGER
)※テーブルはSQLCallコンポーネント（実行する処理：任意のSQLを実行）で作成しました。
********************************************
フロー全体図
メインフロー（OrderImport）
処理の流れは次のとおりです。
・FileGetでCSVを読み込み、RecordLoopで1行ずつ処理します。
・Mapperで顧客コード・金額・注文IDをフロー変数に入れます。
・SwitchRegexpで顧客コードの形式をチェックし、形式が誤っている行はFilePutでエラーファイルに追記します。
・BranchStartで金額をチェックし、上限以内の行はMapperでRecordに変換してRDBPutで登録します。上限を超えた行はExceptionで意図的にエラーを発生させます。
・2つの経路はBranchEnd（合流）でまとめ、EndResponseで終了します。
エラー処理用に、次の3つのフローを同じプロジェクトに作成しました。
・ErrFileNotFound：入力ファイルが無いときに、メールで通知して終了する
・ErrBusinessRule：フロー全体の汎用エラー処理。業務ルール違反をメールで通知して終了する
・ErrInsertFailed：DB登録エラーをSlackで通知し、呼び出し元に戻って処理を続ける
プロジェクトのフロー構成
フロー変数
ループの中で値を判定に使うため、次のフロー変数を作成します。
・customerCode：String
・amount：Integer
・orderId：String
※分岐系のコンポーネントの「評価する値」や「条件式」から、ストリームのフィールドを直接は参照できませんでした。そのため、一度Mapperでフロー変数に入れてから判定しています。
********************************************
FileGetコンポーネント
FileGet1の設定
・ファイルパス：orders_test1.csv
・相対パスの起点：ホームフォルダー
・エラー処理（ファイルが存在しない）：ErrFileNotFound
・ストリーム型：CSV
・読込み開始行：2
・フィールド数：5（order_id、customer_code、item_code、qty、amount）
※「ファイルが存在しない」は、FileGet専用のエラー処理です。ここにエラー処理フローを指定すると、ファイルが無いときだけそのフローに遷移します（パターン2）。
RecordLoopコンポーネント
レコードを1件ずつループします。設定は初期値のままです。
Mapperコンポーネント（Mapper1）
Mapper1のマッピング
・入力をそのまま出力：はい
・マッピング：order_id → orderId、customer_code → customerCode、amount → amount（いずれもフロー変数）
※「入力をそのまま出力」を「はい」にすると、レコードはそのまま後続へ流れます。
SwitchRegexpコンポーネント（パターン1：入力チェックで弾く）
SwitchRegexp1の分岐設定
・評価する値：${flow.customerCode}
・分岐（ラベル：valid）：^C[0-9]{3}$
正規表現に一致した行は「valid」の出力へ進み、一致しなかった行は「デフォルト」の出力へ進みます。
※各フィールドを1つのMapperでまとめてチェックしたい場合は、Mapperの正規表現関数を使う方法もあります（参考の「RegexpMatch」の記事）。
FilePutコンポーネント（エラー行の退避）
・ファイルパス：error_rows.csv
・相対パスの起点：ホームフォルダー
・書込み処理：追加
※「追加」にすると、エラー行が1行ずつ同じファイルに追記されます。
BranchStartコンポーネント（金額のチェック）
BranchStart1の条件式
・条件式：$flow.amount <= 1000000
・判定方法：1つ以上一致
条件が成立した場合は右方向、成立しなかった場合は下方向に出力されます。右をMapper2（登録）、下をException（業務ルール違反）につなぎます。
Exceptionコンポーネント（パターン4：業務ルール違反をわざとエラーにする）
・エラーメッセージ：金額上限超え（注文ID：${flow.orderId}、金額：${flow.amount}）
Exceptionで発生させたエラーは、フローの「汎用エラー処理」に指定したフローへ遷移します。そこで、OrderImportフローのプロパティを次のように設定します。
・汎用エラー処理：ErrBusinessRule
Mapperコンポーネント（Mapper2）
CSVのフィールドを、RDBPutの入力（Record）に対応付けます。
・マッピング：order_id → ORDER_ID、customer_code → CUSTOMER_CODE、item_code → ITEM_CODE、qty → QTY、amount → AMOUNT
※Mapperの後ろにRDBPutを置くと、Mapperの出力ストリームをRecordに変換するか確認されるので「はい」を選びます。
RDBPutコンポーネント（パターン3：通知して処理を続ける）
RDBPut1の設定
・コネクション名：asteria_internal
・実行する処理：Insert
・テーブル名：D1_ORDERS（右クリック→「テーブルの選択」で選び、全フィールドを選択）
・エラー処理（汎用）：ErrInsertFailed
BranchEnd（合流）コンポーネントとEndResponseコンポーネント
FilePutとRDBPutの出力をBranchEndで合流させてから、1つのEndResponseにつなぎます。
・EndResponseの出力形態：最後のストリーム
********************************************
エラー処理フロー：ErrFileNotFound（パターン2：例外を捕まえて通知し、止める）
ErrFileNotFoundフロー
Start → Mapper → SimpleMail → EndResponse の流れです。Mapperではconst関数で本文を作ります。
・本文（const）：入力ファイルが見つからないため、注文データの取込を中止しました。ファイルの配置を確認してください。
SimpleMailは次のように設定します。
・コネクション名：（SMTPコネクション）
・差出人：notify@example.co.jp
・宛先：admin@example.com
・件名：【注文取込】入力ファイルが見つかりません
・入力ストリームを本文とする：はい
※エラー処理フローがEndResponseで終わると、呼び出し元のフローもそこで終了します。
エラー処理フロー：ErrBusinessRule（パターン4の通知）
ErrFileNotFoundを複製し（フローを右クリック→「複製」）、件名と本文だけを変えました。
・件名：【注文取込】業務ルール違反のため取込を中止しました
・本文（const）：金額が上限（1,000,000円）を超える注文があったため、取込を中止しました。入力ファイルを確認してください。
エラー処理フロー：ErrInsertFailed（パターン3の通知と復帰）
ErrInsertFailedフロー
Start → SlackPost → Mapper → ExceptionReturn の流れです。
SlackPostは次のように設定します。Slackコネクションは、FSMC→設定→コネクション→Slackで、SlackアプリのBot User OAuth Token（スコープ：chat:write）を使って作成しています。
・コネクションを使用：はい
・コネクション：（Slackコネクション。チャンネル：#asteria-test）
・メッセージ：【注文取込】DB登録に失敗した注文がありました。該当の1件をスキップして、残りの取込を続行しています。
※投稿先のチャンネルには、あらかじめSlackアプリを追加（/invite）しておきます。
Mapperは、出力のストリーム型をRecordにし、RDBPutと同じフィールド（ORDER_ID、CUSTOMER_CODE、ITEM_CODE、QTY、AMOUNT）を定義しておきます。理由は「使ってみて気になったところ」で説明します。
ExceptionReturnは次のように設定します。
・エラー処理後の動作：次のコンポーネント
※ExceptionReturnは、EndResponseを右クリック→「終了コンポーネントの置き換え」→「ExceptionReturn」で置き換えて配置しました。
********************************************
フローの実行結果
テストデータ（orders_test1.csv）で実行すると、フローは正常終了し、次のようになりました。
・DBに登録された行：O0001、O0002、O0006
・error_rows.csvに退避された行：O0003（顧客コード X99）
・4行目のO0002：DB登録エラー（一意制約違反）→ErrInsertFailedでSlackに通知 → 次の行（O0006）へ進む
error_rows.csvには、形式エラーの行が次のように書き出されます。
"O0003","X99","P-300","1","2000"※FilePutの「書込み処理：追加」は、実行のたびに同じファイルに追記されます。今回もテストで何度か実行したため、同じ行が実行回数分たまっていました。運用では、ファイル名に日付を入れる、取込前にファイルを退避する、などの工夫が必要です。
登録されたデータ
Slackへの通知（重複エラー1件につき1件の投稿）
FileGetのファイルパスを存在しないファイル名（orders_missing.csv）に変えて実行すると、ErrFileNotFoundに遷移し、次のメールが届きました。
入力ファイルが無いときの通知メール
金額1,500,000円の行を加えたファイルで実行すると、Exceptionから汎用エラー処理のErrBusinessRuleに遷移し、次のメールが届きました。
業務ルール違反の通知メール
********************************************
使ってみて気になったところ
1．エラー処理フローをEndResponseで終えると、処理が続かない
最初はErrInsertFailedをStart→EndResponseだけで作りました。その状態で実行すると、重複エラーの行でエラー処理フローに入ったところでフロー全体が終了し、O0006が登録されませんでした。実行結果は「正常終了」と表示されるので、気づきにくい点です。
処理を続けたい場合は、エラー処理フローの最後をExceptionReturnにし、「エラー処理後の動作」を「次のコンポーネント」にします。
2．ExceptionReturnで返すストリームの型を合わせる
SlackPostの直後にExceptionReturnを置くと、次のエラーになりました。
エラー処理フローから返されたストリームの型が違います : [Flow=/DEMO.BlogD1.ErrInsertFailed][Stream=MIME]SlackPostの出力はMIMEです。エラーが起きたRDBPutの出力はRecordなので、ExceptionReturnでもRecordを返す必要があります。SlackPostの後ろに、出力をRecordにしたMapperを置いて解決しました。
なお、このエラーが出ていた段階や試行錯誤の途中では、同じ通知がSlackに複数回投稿されていました。修正後のフローで実行したときは、重複エラー1件につき1件の投稿でした。複数回投稿された原因は、今後調査して別の記事でご紹介する予定です。
3．ループ内で終了コンポーネントを分けると「フローが複数箇所から結果を返しています」
FilePutの後ろとRDBPutの後ろにそれぞれEndResponseを置いたところ、このエラーで異常終了しました。BranchEnd（合流）でまとめて1つのEndResponseにつなぐと解消しました。
さらに、EndResponseの出力形態が「ストリームをまとめる」のままだと「パックするストリームのストリーム型が異なります」になりました。FilePut側（CSV）とRDBPut側（Record）で型が違うためです。出力形態を「最後のストリーム」に変えて解消しました。
4．BranchStartの条件式の書き方
条件式に「flow.amount > 1000000」と書くと、コンパイル時に「条件式が不正です」になりました。「${flow.amount}」と書くとコンパイルは通りましたが、意図した方向に分岐しませんでした。
BranchStartの条件式はRQLで書き、フロー変数は「$flow.amount」のように参照します。このように直して、期待どおりに分岐しました。また、条件が成立したときは右、成立しなかったときは下に出力されます。
********************************************
まとめ
エラーの種類ごとに「止めるか、続けるか」と「通知先」を先に決めておくと、フローの中で使うコンポーネントも自然に決まります。止めるエラーはエラー処理フローで通知してから終了し、続けるエラーはExceptionReturnで呼び出し元に戻ります。入力チェックで弾ける誤りは、エラーにせず分岐で退避します。ExceptionReturnで返すストリームの型と、ループ内の終了コンポーネントのまとめ方には注意が必要でした。
最後までお読みいただき、ありがとうございました。
少しでもご参考になれば幸いです。
参考：
・Asteria Warpフローデザイナー：意図的にエラーを発生させるExceptionと、エラー処理フローから復帰するExceptionReturn
・RegexpMatch（正規表現マッチ）によるフォーマットエラー検出方法
・ASTERIA Warpにおけるメール送信「SimpleMailコンポーネント、SMTPコンポーネント＆MIMEEncodeコンポーネント」
・Asteria WarpフローデザイナーとSlackその1：SlackPost/SlackGetを使用してSlackとメッセージの送信/取得を行う
・BranchStart（ASTERIA Warp ヘルプ）