本記事は
flaskを使ってRestAPIサーバを作ってみるの「作成」について記述しています。
ここまでの流れはflaskを使ってRestAPIサーバを作ってみるを参照してください。
作成編で説明するコードは feature/chapter2-create ブランチにあります。
flask_sv プロジェクトのディレクトリ配下で次のコマンドを実行してください。
$ git checkout feature/chapter2-create
2. 作成
参照の次に、Accountレコードを作るApiを考えましょう。
入力には作られるAccountオブジェクトで必須となる情報を受け取れる様にする必要があります。
Accountオブジェクトの場合、以下の様な情報で構成されています。
| 項目名 | カラム名 | データ型 |
|---|---|---|
| アカウントID | id | 数値 |
| アカウント名称 | account_name | 文字列 |
| 有効開始日 | start_on | 時間 |
| 有効終了日 | end_on | 時間 |
| 作成者 | created_by | アカウントを特定するID |
| 作成日時 | created_at | 時間 |
| 更新者 | updated_by | アカウントを特定するID |
| 更新日時 | updated_at | 時間 |
| ステータス | status | 業務内で定義 |
これはテーブル定義としては次の様なスキーマで作られているということをFlask+MySQL on dockerで話しています。
mysql> desc account;
+--------------+-------------+------+-----+---------+----------------+
| Field | Type | Null | Key | Default | Extra |
+--------------+-------------+------+-----+---------+----------------+
| id | int | NO | PRI | NULL | auto_increment |
| account_name | varchar(64) | NO | UNI | NULL | |
| start_on | datetime | NO | | NULL | |
| end_on | datetime | NO | | NULL | |
| created_by | int | YES | | NULL | |
| created_at | datetime | YES | | NULL | |
| updated_by | int | YES | | NULL | |
| updated_at | datetime | YES | | NULL | |
| status | int | NO | | NULL | |
+--------------+-------------+------+-----+---------+----------------+
9 rows in set (0.08 sec)
mysql>
作成時に必須となるのは更新者、更新日時を除いた全項目です。
しかし、アカウントID（自動インクリメント）、作成日時(作られた時のシステム時間)、ステータス(作成時は固定値)は、システムの中で決められます。
そのためAPIで受け取るべきなのはアカウント名称と作成者（ログイン中のユーザー）、有効開始日、有効終了日です。
作成者もシステム中で決められます、ログイン時にクッキーやセッションに登録するということが可能と思えますが、今はAPIのリクエストで指定することにしましょう。
受信すべきリクエストは次の様になります。
createのリクエスト
createリクエストのJSON例です。
{
"account_name": "sato",
"start_on": "2021-01-01 00:00:00",
"end_on": "2021-12-31 23:59:59",
"opration_account_id": 123
}
2.1. モデル層・エンティティ
作成のモデル層の実装は次のコードです。
def create(account_dict, operation_account_id):
account = Account()
account.account_name = account_dict['account_name']
account.start_on = account_dict['start_on']
account.end_on = account_dict['end_on']
account.created_by = account_dict['created_by']
account.created_at = account_dict['created_at']
account.updated_by = account_dict['updated_by']
account.updated_at = account_dict['updated_at']
account.status = account_dict['status']
Session = sessionmaker(bind=engine)
ses = Session()
ses.begin()
try:
ses.add(account)
ses.commit()
res = True
except:
ses.rollback()
res = False
finally:
ses.close()
return res
コードの解説をします。
受け取ったaccount_dictからAccountオブジェクトを作っています。
システムで設定できる項目として挙げた|created_by(作成者)、created_at(作成日時)、updated_by（更新者）、updated_at（更新日時）、status（ステータス）はモデル層で設定するのでなく業務要件でその値が設定されると考えてドメイン層に寄せましょう。
def create(account_dict, operation_account_id):
account = Account()
account.account_name = account_dict['account_name']
account.start_on = account_dict['start_on']
account.end_on = account_dict['end_on']
account.created_by = account_dict['created_by']
account.created_at = account_dict['created_at']
account.updated_by = account_dict['updated_by']
account.updated_at = account_dict['updated_at']
account.status = account_dict['status']
DBとのセッションを作成し、トランザクションを開始します。
更新なので更新の途中での失敗を想定した実装にします。
Session = sessionmaker(bind=engine)
ses = Session()
ses.begin()
Accountオブジェクトを変更してセッションにaddすることでDBに反映さレます。
正常に更新されたら次のステップのcommitが実行されます。
try:
ses.add(account)
ses.commit()
res = True
もし、exceptionが発生したらexceptブロックでrollbackが行われます。
except:
ses.rollback()
res = False
finally:
ses.close()
return res
参照のgetByIdのところで、Exceptionが起きたときの対処など必要と書きましたが
これが実現方法です。
ただこの書き方だとExceptionのメッセージが残らないので障害の原因がわからないです。
2.2. ドメイン層・ビジネスロジック
ドメイン層の実装は次のコードです。
def create(account_request, operation_account_id):
"""
/account/createで呼び出されたAPIの検索処理
Parameters
----------
account_request : json
作成するアカウント詳細
operation_account_id : int
Webアプリケーション操作アカウントのID
Returns
-------
JSON形式の処理結果
正常
異常
"""
account = {
'account_name' : str(account_request['account_name']),
'start_on' : str(account_request['start_on']),
'end_on' : str(account_request['end_on']),
'created_by' : operation_account_id,
'created_at' : strftime(datetime.datetime.now()),
'updated_by' : operation_account_id,
'updated_at' : strftime(datetime.datetime.now()),
'status' : Status.getStatusKey("NEW")
}
try:
if Account.create(account, operation_account_id) == True:
code="I0001"
message="Created Account Succesfuly."
else:
code="E0001"
message=""
except:
code="E0009"
message="Created failed"
result_json = {
"body": "",
"status": {
"code" : code,
"message" : message,
"detail" : ""
}
}
return result_json
コードの解説をします。
APIから受け取ったJSON構造のリクエストから各項目を抜き出してAccountのcreate関数に渡すDICTを作ります。
account = {
'account_name' : str(account_request['account_name']),
'start_on' : str(account_request['start_on']),
'end_on' : str(account_request['end_on']),
'created_by' : operation_account_id,
'created_at' : datetime.datetime.now(),
'updated_by' : operation_account_id,
'updated_at' : datetime.datetime.now(),
'status' : Status.getStatusKey("NEW")
}
Accountのcreateを実行を実行して、その結果から正常か異常かの返却情報を作成します。
try:
if Account.create(account, operation_account_id) == True:
code="I0001"
message="Created Account Succesfuly."
else:
code="E0001"
message="Created Account Failed."
こちらはAccount作成時のエラー以外が発生した場合。
except:
code="E0009"
message="Created failed"
作成のAPIなので結果のみ返却すれば良いのでbodyは空にしています。
設計としては"body"をとってしまう、"status"の項目名なしで1回層減らすというのも考えられますが、受け取り側の実装で同じフォーマットの方が処理しやすいでしょう。
result_json = {
"body": "",
"status": {
"code" : code,
"message" : message,
"detail" : ""
}
}
return result_json
課題 2.1
create APIは作成成功したか失敗したかどうかはAPI呼び出し側から確認することができますが、失敗した時に何が原因で失敗したのかわかりません。全てのエラーをメッセージに埋め込める様に変更したAPI(/account/create_with_message)を作成してください。
課題 2.2
account_nameは、ログインに使うことを想定していることにします。ログインに使う場合マルチバイト文字では不便なので、リクエスト時に英数と一部の記号("-", "_")のみ許される様にチェックを入れてください。API(/account/create_with_check)として作成してください。
2.3. アプリケーション層・フロントへのIF
HTTPリクエストでアクセスされるURLと処理のマッピング、レスポンスをJSON形式で返却する処理を記載してあります。
http://サーバーのIP/account/create にPOSTでアクセスされたらAccountApi.createを実行して結果をJSON形式で返却します。
@api_bp.route('/account/create', methods=['POST'])
def createAccount():
payload = request.json
response_json = AccountApi.create(payload, system_account_id)
return jsonify(response_json)
2.4. テストコードからの実行
テストコード全文
def test_account_create():
"""
"""
# modelから試験データ登録
test_account_name = 'api_account_get'
test_start_on = '2021-06-23 00:00:00'
test_end_on = '2030-12-31 00:00:00'
payload = {
'account_name' : test_account_name,
'start_on' : test_start_on,
'end_on' : test_end_on
}
# createのテスト
# APIの実行
url = f"http://localhost:5000/api/account/create"
headers = {'Accept-Encoding': 'identity, deflate, compress, gzip',
'Accept': '*/*', 'User-Agent': 'flask_sv/0.0.1',
'Content-type': 'application/json; charset=utf-8',
}
response = requests.post(url, headers=headers)
assert response.status_code == 200
data = json.loads(response.text)
assert data['body'] == ""
assert data['status']['code'] == "I0001"
assert data['status']['message'] == "Created Account Succesfuly."
# 作成されたデータの確認
account_dict = {
'account_name' : test_account_name
}
result = Account.search(account_dict, 999)
account_id = result[0].id
result_json = AccountApi.getById(account_id, 100)
assert result_json['body']['name'] == "account"
assert result_json['body']['account_name'] == test_account_name
assert result_json['body']['start_on'] == test_start_on
assert result_json['body']['end_on'] == test_end_on
assert result_json['status']['code'] == "I0001"
assert result_json['status']['message'] == ""
コードの解説をします。
テストAccountを作ります。その元になるパラメータです。
# modelから試験データ登録
test_account_name = 'api_account_get'
test_start_on = '2021-06-23 00:00:00'
test_end_on = '2030-12-31 00:00:00'
payload = {
'account_name' : test_account_name,
'start_on' : test_start_on,
'end_on' : test_end_on
}
createのAPIからPOSTでリクエストを投げて作成します。
# createのテスト
# APIの実行
url = f"http://localhost:5000/api/account/create"
headers = {'Accept-Encoding': 'identity, deflate, compress, gzip',
'Accept': '*/*', 'User-Agent': 'flask_sv/0.0.1',
'Content-type': 'application/json; charset=utf-8',
}
response = requests.post(url, headers=headers)
APIの処理結果を確認します。
assert response.status_code == 200
data = json.loads(response.text)
assert data['body'] == ""
assert data['status']['code'] == "I0001"
assert data['status']['message'] == "Created Account Succesfuly."
APIの処理結果だけでは登録された内容が正しいか分からないのでAccountにて検索し、アカウントIDを取得します。
検索はまだ出てきていませんが3.検索で説明します。
# 作成されたデータの確認
account_dict = {
'account_name' : test_account_name
}
result = Account.search(account_dict, 999)
account_id = result[0].id
取得したアカウントIDをgetByIdから検索し、目的のレコードと比較します。
result_json = AccountApi.getById(account_id, 100)
assert result_json['body']['name'] == "account"
assert result_json['body']['account_name'] == test_account_name
assert result_json['body']['start_on'] == test_start_on
assert result_json['body']['end_on'] == test_end_on
assert result_json['status']['code'] == "I0001"
assert result_json['status']['message'] == ""