4 月の入社に合わせて 30 人分のユーザを登録し、同じ 4 月の異動で 50 人の所属を付け替えることになったとします。この記事では、人事システムから出力した CSV をもとに、この 2 つを REST API で反映する方法を説明します。
ユーザの登録は、［システム設定］＞［ユーザ一覧］の［ユーザ一括登録］でもできます。1 回かぎりであれば、そちらのほうが手軽です。手順は「ユーザを一括登録する」で説明しています。
REST API を使うのは、この作業が毎年・毎月くり返される場合です。人事システムの出力をそのまま流し込めるようにしておけば、次回からは CSV を差し替えて実行するだけで済みます。
この記事では、組織そのもの（部や課の新設、名称の変更）は変わらないものとして説明します。また、人事システム側の名前を「部署名」、ワークフロー基盤側を「組織」と呼び分けます。
コード例は Python 3 で、HTTP リクエストの送信に Requests パッケージを使います。コードは節ごとに分けて説明し、末尾の「コード全体」に 1 本にまとめています。
用意する CSV
入社する人と異動する人は、まとめて 1 つの CSV にします。人事システムから、次の形で出力し、april.csv として保存します。
name,email,locale,timezone,qgroup1,qgroup2
田中 太郎,taro.tanaka@example.com,,,営業部,
鈴木 花子,hanako.suzuki@example.com,en,America/Los_Angeles,開発部,品質保証室name がユーザ名、email がメールアドレスです。入社する人はこの内容で登録し、すでに登録されている人は、書かれた部署名の組織へ所属を付け替えます。
列は、API のパラメータ名に合わせた列と、部署名の列の 2 種類です。 name email locale timezone が API のパラメータ名に合わせた列です。列名をそろえておくと、CSV の 1 行を読み込んだ辞書を、そのまま API へ渡せます。qgroup1 qgroup2 が部署名の列で、API のパラメータにはありません。コードの中で組織 ID に変換してから使います。残った row をそのまま API に渡すので、人事システムの出力に社員番号などほかの列がある場合は、この 6 列だけに絞っておきます。
部署名の列は、兼務の数だけ並べます。 先頭の qgroup1 が、主として属する組織です。上の例では、鈴木 花子が開発部と品質保証室の兼務で、主として属するのは開発部です。画面から一括登録するときの CSV と同じ並べ方です。兼務がない人の 2 列目は空欄のままで構いません。
locale と timezone は、空欄ならシステムの既定が使われます。海外拠点の社員など、個別に指定したい人の行にだけ値を入れます。この 2 列はユーザの登録にだけ使われるため、すでに登録されている人の行に書いても、その人の設定は変わりません。
locale に指定できる値は ja en ko zh_TW zh_CN です。timezone に指定できる値は Asia/Tokyo のような地域名のタイムゾーン ID で、［アカウント設定］＞［タイムゾーン］の選択肢に載っているものに限られます。
API を呼び出す準備
実行するユーザには、ユーザ管理権限と、Basic 認証による API アクセスの許可が必要です。 ［システム設定］＞［ユーザ一覧］で対象のユーザを開き、ユーザ管理権限があることを確認します。次に、［Basic 認証による API アクセス］の行の［編集］から「許可」にチェックを入れてください。認証に使うパスワードは、［アカウント設定］＞［パスワード］の［API パスワード］タブに表示されている文字列です。ログインパスワードとは別のものです。
接続先と認証情報を、次のように用意しておきます。
import csv
import requests
from requests.auth import HTTPBasicAuth
BASE = 'https://example.questetra.net'
AUTH = HTTPBasicAuth('admin@example.com', '{API パスワード}')
API の呼び出しは、次の関数を通します。 失敗したときは、応答本文を出力してから止めます。本文には errorCode のほか、どのパラメータ（param）にどの値（input）を渡したかが入っています。エラーコードの一覧は「R3175: 応答エラーリスト (ワークフロー基盤 API)」にあります。
def check(r):
if not r.ok:
print(r.status_code, r.text)
r.raise_for_status()
return r
def call(method, path, **kwargs):
return check(requests.request(method, f'{BASE}{path}', auth=AUTH, **kwargs))パラメータの一覧や応答の形式は、ワークフロー基盤の左メニューの［REST API Reference］から開く Swagger で確認できます。使い方は「SwaggerでQuestetra REST API を使ってみる」で説明しています。
組織一覧の取得
ユーザの登録にも所属の付け替えにも、API が使うのは組織の ID です。 CSV にあるのは部署名なので、先に組織の一覧を取得して、組織名から ID を引く辞書を作っておきます。以降の処理は、この辞書を通して組織 ID を得ます。
一覧を返す API は、1 回の呼び出しで返る件数に上限があります。limit で 1 回あたりの件数を指定し、start をずらしながら、返ってきた件数が limit より少なくなるまでくり返します。
LIMIT = 10
groups = {}
start = 0
while True:
r = call('GET', '/API/User/Qgroup/list',
params={'start': start, 'limit': LIMIT})
qgroups = r.json()['qgroups']
groups.update({g['name']: g['id'] for g in qgroups})
if len(qgroups) < LIMIT:
break
start += LIMIT
limit には 1000 まで指定できます。この例では、くり返しの動きが分かるよう 10 にしています。1000 にしても、組織の数がそれを超える場合に備えて、くり返す形は残しておきます。
異動：所属の付け替え
ユーザの所属を設定する API は /API/UGA/Quser/setMembership です。memberships に指定した内容で、そのユーザの所属が置き換わります。異動先だけを指定すれば、前の所属は残りません。関数にしておきます。
def move_user(quser_id, memberships):
call('POST', '/API/UGA/Quser/setMembership',
json={'quserId': quser_id, 'memberships': memberships})
memberships には所属する組織を 1 つずつ並べ、主として属する組織に primary を付けます。 開発部（ID 12）を主として、品質保証室（ID 34）を兼務する場合は [{'qgroupId': 12, 'primary': True}, {'qgroupId': 34}] です。primary は 1 つだけ指定してください。CSV の部署名からこの形に組み立てる処理は、「1 行ごとの前処理と振り分け」で説明します。
異動では、ユーザ名を変更しません。 この関数が扱うのは所属だけなので、CSV の name は登録のときにしか使われません。改姓などでユーザ名を変える場合は、/API/UGA/Quser/update に id と name を渡します。
入社：ユーザの登録
ユーザを追加する API は /API/UGA/Quser/add です。渡す row は、CSV の 1 行から部署名の列を外し、空欄の列を落としたものです。この前処理は「1 行ごとの前処理と振り分け」で行います。
def add_user(row, memberships):
if len(memberships) == 1:
row['primaryQgroupId'] = memberships[0]['qgroupId']
call('POST', '/API/UGA/Quser/add', data=row)
return
r = call('POST', '/API/UGA/Quser/add', data=row)
move_user(r.json()['quser']['id'], memberships)
兼務がなければ、登録のときに所属も決まります。 add の primaryQgroupId に組織 ID を渡せば、主として属する組織が設定されるためです。兼務がある人は、primaryQgroupId では 1 つしか指定できないので、登録したあとに前の節の move_user を呼びます。
パスワードは指定しません。 password を指定しないと、パスワードが未設定の状態でユーザが作られます。各ユーザには、ログインページの「パスワードを忘れた場合はこちら」から自分でパスワードを設定してもらいます。この流れは画面から一括登録した場合と同じで、「ユーザを一括登録する」で説明しています。
1 行ごとの前処理と振り分け
CSV を 1 行ずつ読み、部署名を組織 ID に変換してから、メールアドレスでユーザを探します。見つかれば move_user、見つからなければ add_user を呼びます。
with open('april.csv', encoding='utf-8') as f:
for row in csv.DictReader(f):
names = [row.pop(k) for k in list(row) if k.startswith('qgroup')]
names = [n for n in names if n]
row = {k: v for k, v in row.items() if v}
if not names:
print(row.get('email'), '部署名が空です')
continue
missing = [n for n in names if n not in groups]
if missing:
print(row.get('email'), '組織一覧にない部署名:', missing)
continue
memberships = [{'qgroupId': groups[n]} for n in names]
memberships[0]['primary'] = True
found = requests.get(f'{BASE}/API/User/Quser/find', auth=AUTH,
params={'email': row['email']})
if found.status_code != 400:
check(found)
if found.ok:
move_user(found.json()['quser']['id'], memberships)
print(row['email'], '所属を変更しました')
else:
add_user(row, memberships)
print(row['email'], '登録しました')
空欄の列は、API に渡さずに落とします。 部署名の列を row から取り出しているので、残った row は API のパラメータだけです。空文字のまま渡すと、その値を指定したことになり、locale や timezone はエラーになります。
部署名の列がすべて空の行と、部署名が組織一覧にない行は、メールアドレスを出力して次の行に進みます。 所属が空のまま move_user を呼ぶと、その人の所属がすべて外れてしまいます。部署名が組織一覧にない行は、人事システムとワークフロー基盤とで表記が食い違っている行です。表記ゆれはここで見つかります。
memberships は、部署名を組織 ID に置き換えて並べたものです。 CSV の qgroup1 が先頭になるので、主として属する組織は qgroup1 の部署になります。
find は、未登録のメールアドレスには 400 を返します。 それだけを「見つからない」として扱い、認証エラーなどほかのエラーは check で止めます。call を通さないのはそのためです。
途中で失敗しても、CSV を先頭から流し直せます。 メールアドレスで振り分けているため、登録済みの行は所属の設定をやり直すだけになり、同じユーザが二重に登録されることはありません。
コード全体
ここまでのコードを 1 本にまとめたものです。実行する前に、次の「実行の進め方」を読んでください。
import csv
import requests
from requests.auth import HTTPBasicAuth
BASE = 'https://example.questetra.net'
AUTH = HTTPBasicAuth('admin@example.com', '{API パスワード}')
LIMIT = 10
def check(r):
if not r.ok:
print(r.status_code, r.text)
r.raise_for_status()
return r
def call(method, path, **kwargs):
return check(requests.request(method, f'{BASE}{path}', auth=AUTH, **kwargs))
def move_user(quser_id, memberships):
call('POST', '/API/UGA/Quser/setMembership',
json={'quserId': quser_id, 'memberships': memberships})
def add_user(row, memberships):
if len(memberships) == 1:
row['primaryQgroupId'] = memberships[0]['qgroupId']
call('POST', '/API/UGA/Quser/add', data=row)
return
r = call('POST', '/API/UGA/Quser/add', data=row)
move_user(r.json()['quser']['id'], memberships)
groups = {}
start = 0
while True:
r = call('GET', '/API/User/Qgroup/list',
params={'start': start, 'limit': LIMIT})
qgroups = r.json()['qgroups']
groups.update({g['name']: g['id'] for g in qgroups})
if len(qgroups) < LIMIT:
break
start += LIMIT
with open('april.csv', encoding='utf-8') as f:
for row in csv.DictReader(f):
names = [row.pop(k) for k in list(row) if k.startswith('qgroup')]
names = [n for n in names if n]
row = {k: v for k, v in row.items() if v}
if not names:
print(row.get('email'), '部署名が空です')
continue
missing = [n for n in names if n not in groups]
if missing:
print(row.get('email'), '組織一覧にない部署名:', missing)
continue
memberships = [{'qgroupId': groups[n]} for n in names]
memberships[0]['primary'] = True
found = requests.get(f'{BASE}/API/User/Quser/find', auth=AUTH,
params={'email': row['email']})
if found.status_code != 400:
check(found)
if found.ok:
move_user(found.json()['quser']['id'], memberships)
print(row['email'], '所属を変更しました')
else:
add_user(row, memberships)
print(row['email'], '登録しました')
実行の進め方
まず 1 行だけの CSV で実行し、［システム設定］＞［ユーザ一覧］で、ユーザ名・メールアドレス・所属組織が意図どおりかを確認してください。 POST の API は、ワークフロー基盤のデータをその場で書き換えます。入社の行と異動の行を 1 行ずつ試すと、振り分けの動きもあわせて確かめられます。確認できてから、残りの行を流します。
登録したユーザにパスワード設定を案内するタイミングも、あわせて決めておきます。登録した時点でユーザは有効になるため、入社日より前に登録する場合は、案内を入社日に合わせて送ることになります。
