n8n.io で DBへのupsertを実現するフロー

## はじめに

n8n.ioは簡単にフロー処理を実現できるOSSです．
本記事ではn8n.ioのVersion 0.83.0のdockerイメージを用いて，WebhookにポストされたJSONデータをDBに登録するフローを構築しました． 
その際，DBに重複したデータがある場合はUpdate,重複が無いデータをInsertするフローを作成したので，その方法を記載します．

## フロー概略

ざっくり書くとJsonで受け取ったデータを全てUpdateをかけた後に，Findで見つけたデータを除外して残ったデータをInsertするという流れです，
PostgreSQLフローで書きましたが基本的にどのDBフローに対しても使用できる構成になっています．
DBのフローはDBにデータがなくてもUpdateではErrorを吐きませんが，Insertではデータが存在した場合Errorで中断されるので，このような構成のフローになっています．

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/850127/07581570-76c0-059d-e86c-96af123c27c5.png)

## Webhookの設定

以下のようにBasic認証の設定でWebhookのTriggerを設定します．
Credentialsの欄は既存の認証を用いるか，Create Newで新しいものを設定してください．
さらにAdd OptionsからRaw Bodyを選択して有効化します
また，Webhook URLsをクリックして拡張し，TestのWebhooks URLをコピーしておきます．

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/850127/3a3c19b3-3b58-66de-ca29-748be4887f02.png)


## 入力の仕様

以下のように,DBに登録したいデータがvalues以下のリストに格納されたデータとします．

```json
{"values":[
 {"id": 1, "name": "hoge"},
 {"id": 2, "name": "fuga"}
]}
```

この用に記述されたJSONを以下のコマンドでn8nのWebhookに送信します．

```bash
curl --netrc-file <credential file> -X POST <コピーしたWebhookのURL> -d @<json file>
```

credential fileの書き方は以下のように認証情報を記載します．

```
machine <n8nのホスト名> login <user> password <password>
```

## JSONの抽出

Move Binary DataフローをWebhookの後ろに配置して以下のように設定します．

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/850127/2216ff9b-d379-e80e-6049-626d90225281.png)

## データリスト化

JSONからDBに登録するデータを抽出するためにMove Binary Dataの後ろにFunctionノードを配置して処理を記述します

```js
var newList = [];
for ( const data of items[0].json.values){
  newList.push({json: data})
}
return newList;
```

## Updateフロー

データリスト化の後ろにUpdateフローを追加します．
DBの認証情報や，入力したいテーブルの情報は適宜入力してください．
Postgres以外でもUpdateというのがあると思いますのでこちらも適宜読み替えてください．

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/850127/0638e8ca-c5cd-e34c-b4f5-f851a9912a41.png)

## Idリスト化

Binary Move Dataから更に分岐させてFunctionItemフローを追加します．
このFunctionではJsonデータの中身から検索キーとなるidをリスト化します．

```js
idList = []
for ( const i of item.values){
  idList.push(i.id)
}
return {"ids": idList};
```

## Findフロー

Idリスト化のフローの後ろにDBのフローを接続し，Findもしくはクエリを使用するものを選択します．

Postgresの場合は,認証情報を記載し，OperationはExecute Queryとします．
そしてQueryをParameter OptionsからAdd Expressionを選択肢，以下のように記述します．ただしtable nameは適宜自身の環境に読み替えて記載してください．

```sql
SELECT * FROM <table name> WHERE id IN ('{{$node["FunctionItem"].json["ids"].join("','")}}')
```

## merge

そして２つのDBフローの後ろにMergeフローを接続します．
Mergeフローでは以下のように設定をします．
これにより，Updateに入力したデータと，Findで見つけたデータを比較して，idが一致したデータは後段に流れなくなります．

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/850127/b0c2dff8-888b-1d75-aeae-fc6faeec41e9.png)

## Insert

最後にMergeフローの後段にDBフローを追加してInsertを定義します．

接続情報，Tableなどは自身の環境に読み替えてください．

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/850127/14be0e41-fda7-79af-a775-f96b41a00f71.png)

## おわりに

以上でフローの構築は終了です．
保存してフローのWebhookにデータを送ってみてください．
画面上のExecute WorkFlowを実行した場合は，TestのURLにデータを送信すると動作を画面上で確認できます．
右上のActiveトグルボタンを有効にした場合，ProductionのURLが常時待受状態になり，受信できるので本番ではこちらを利用してください．

DBのクエリの知識がなくても，簡単な検索のクエリだけ調べれば，既存のフローを組み合わせるだけでDBへ書き込みができるRestAPIを構築できるので，非常に汎用性が高いと思います．

10フィールドのデータを100個格納したJSONを送信しましたが実行時間も200msec程度なので個人用途でなら十分な性能だと思います．