本記事は
flaskを使ってRestAPIサーバを作ってみるの「検索」について記述しています。
ここまでの流れはflaskを使ってRestAPIサーバを作ってみるを参照してください。
検索編で説明するコードは feature/chapter2-search ブランチにあります。
flask_sv プロジェクトのディレクトリ配下で次のコマンドを実行してください。
$ git checkout feature/chapter2-search
3. 検索
検索も考え方は作成とほぼ同じですが、検索に必要な項目はその時よって異なります。
全ての項目を受け取れるJSONを設計しますが、項目は任意とし、項目がなかったら検索条件に加えないことにします。
また、検索というのは与えられる項目について完全一致、部分一致、範囲検索などができるのが通常のシステムでしょう。今回は限定的に完全一致のみ実施することとします。
受信すべきリクエストは次の様になります。
searchのリクエスト
{
"account_name": "アカウント名",
"start_on": "2021-01-01 00:00:00",
"end_on": "2021-12-31 23:59:59",
"opration_account_id": 123,
"created_by" : 999,
"created_at" : "2021-08-25 00:00:00",
"updated_by" : 999,
"updated_at" : "2021-08-25 00:00:00",
"status" : statusの数値
}
3.1. モデル層・エンティティ
検索のモデル層の実装は次のコードです。
backend/src/model/Account.py
def search(account_dict, operation_account_id):
"""
dictアカウントからaccountテーブルを検索し、該当したAccountオブジェクト群を取得する
Parameters
----------
{
'account_name' : 文字列str(self.account_name),
'start_on' : 文字列 '2020-05-01 00:00:00',
'end_on' : 文字列 '2020-12-31 00:00:00',
'created_by' : account_id,
'created_at' : 文字列 '2020-12-31 00:00:00',
'updated_by' : account_id,
'updated_at' : 文字列 '2020-12-31 00:00:00',
'status' : statusの数値
}
Returns
-------
Accountオブジェクトのリスト
"""
print(f"account_dict={account_dict}")
Session = sessionmaker(bind=engine)
ses = Session()
res = None
rs = ses.query(Account)
v = account_dict.get('account_name')
if (v != None):
rs = rs.filter(Account.account_name==v)
v = account_dict.get('start_on')
if (v != None):
rs = rs.filter(Account.start_on==v)
v = account_dict.get('end_on')
if (v != None):
rs = rs.filter(Account.end_on==v)
v = account_dict.get('created_by')
if (v != None):
rs = rs.filter(Account.created_by==v)
v = account_dict.get('created_at')
if (v != None):
rs = rs.filter(Account.created_at==v)
v = account_dict.get('updated_by')
if (v != None):
rs = rs.filter(Account.updated_by==v)
v = account_dict.get('updated_at')
if (v != None):
rs = rs.filter(Account.updated_at==v)
v = account_dict.get('status')
if (v != None):
rs = rs.filter(Account.status==v)
rs = rs.filter(Account.status!=Status.getStatusKey("DELETE"))
res = rs.all()
lambda r: print(f"r={r}"),res
ses.close()
return res
コードの解説をします。
セッションを開始します。
Account.py
Session = sessionmaker(bind=engine)
ses = Session()
検索する項目をクエリにつなげる作業を１つづつ実施します。
Account.py
res = None
rs = ses.query(Account)
v = account_dict.get('account_name')
if (v != None):
rs = rs.filter(Account.account_name==v)
v = account_dict.get('start_on')
if (v != None):
rs = rs.filter(Account.start_on==v)
v = account_dict.get('end_on')
if (v != None):
rs = rs.filter(Account.end_on==v)
v = account_dict.get('created_by')
if (v != None):
rs = rs.filter(Account.created_by==v)
v = account_dict.get('created_at')
if (v != None):
rs = rs.filter(Account.created_at==v)
v = account_dict.get('updated_by')
if (v != None):
rs = rs.filter(Account.updated_by==v)
v = account_dict.get('updated_at')
if (v != None):
rs = rs.filter(Account.updated_at==v)
v = account_dict.get('status')
if (v != None):
rs = rs.filter(Account.status==v)
検索されてはいけないレコードを排除する条件を設定します。ステータスで削除を表す論理削除を使い場合はスタータスが削除のレコードを除外するクエリを追加しましょう。
Account.py
rs = rs.filter(Account.status!=Status.getStatusKey("DELETE"))
SQLの実行結果、複数のレコードがヒットしても全件返却します。
Account.py
res = rs.all()
ses.close()
return res
3.2. ドメイン層・ビジネスロジック
ドメイン層の実装は次のコードです。
backend/src/restapi/AccountApi.py
def search(account_request, user_id):
"""
/account/searchで呼び出されたAPIの検索処理
Parameters
----------
account_request : json
アカウント検索項目
user_id : int
Webアプリケーション操作アカウントのID
Returns
-------
JSON形式の処理結果
正常
異常
"""
print(f"account_request={account_request}")
try:
results = Account.search(account_request, user_id)
code="I0001"
message=f"Found ({len(results)}) records."
except Exception as e:
code="E0009"
message="Search failed: " + str(e)
result_json = {
"body": list(map(lambda s: s.toJson(), results)),
"status": {
"code" : code,
"message" : message,
"detail" : ""
}
}
return result_json
コードの解説をします。
Accountは、処理結果によってcommitやrollbackする必要がないため、Account.pyの中でエラーハンドリングしなくても済みます。
そのためここでexceptionの処理を実装しています。また、createまでの課題であったエラーメッセージの取得とレスポンスへの埋め込みも実施することで、APIを呼び出しているクライアント側にエラー内容を伝えることができます。
AccountApi.py
try:
results = Account.search(account_request, user_id)
code="I0001"
message=f"Found ({len(results)}) records."
except Exception as e:
code="E0009"
message="Search failed: " + str(e)
JSONに検索結果を埋め込んでいます。検索結果は複数あるためリストのためAccountオブジェクトそれぞれをJSONにする関数を用意し、lambdaで簡潔に記載しています。
AccountApi.py
result_json = {
"body": list(map(lambda s: s.toJson(), results)),
"status": {
"code" : code,
"message" : message,
"detail" : ""
}
}
return result_json
3.3. アプリケーション層・フロントへのIF
HTTPリクエストでアクセスされるURLと処理のマッピング、レスポンスをJSON形式で返却する処理を記載してあります。
http://サーバーのIP/account/search にPOSTでアクセスされたらAccountApi.searchを実行して結果をJSON形式で返却します。
backend/src/api.py
@api_bp.route('/account/search', methods=['POST'])
def searchAccount():
payload = request.json
response_json = AccountApi.search(payload, system_account_id)
return jsonify(response_json)
課題 3.1
検索では、時間の範囲を指定した検索をしたい。それを実現するAPI(/account/search_range)の設計と実装をしてください。
課題 3.2
アカウント名の検索は完全一致でなく部分一致の方が需要がありそうです。部分一致の場合は"*"を前後につけることにします。つまり片方だけつけることで前方一致、後方一致も実現できるわけです。
上記を満たすAPI(/account/search_intermediate)を実装してください。
3.4. テストコードからの実行
テストコード全文
backend/tests/restapi/test_AccountApi.py
def test_account_search():
"""
"""
account = {
'account_name' : "search_account",
'start_on' : '2021-05-23 00:00:00',
'end_on' : '2030-12-31 00:00:00',
'created_by' : 999,
'created_at' : datetime.datetime.now(),
'updated_by' : 999,
'updated_at' : datetime.datetime.now(),
'status' : Status.getStatusKey("NEW")
}
# createのテスト
assert Account.create(account, 999) == True
payload = {
"account_name":"search_account",
"start_on":"2021-05-23 00:00:00",
"end_on":"2030-12-31 00:00:00"
}
#result = Account.search(query, 999)
# APIから確認
url = f"http://localhost:5000/api/account/search"
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, json=payload)
# HTTP Statusコードが200であること
assert response.status_code == 200
print(f"test_account_search():json response.text={response.text}")
# BODYをjsonでパースできること
data = json.loads(response.text)
print(f"test_account_search():json data={data}")
assert data['body'][0]['account_name'] == payload["account_name"]
assert data['body'][0]['start_on'] == payload["start_on"]
assert data['body'][0]['end_on'] == payload["end_on"]
assert data['body'][0]['created_by'] == 999
assert data['status']['code'] == 'I0001'
assert data['status']['message'] == 'Found (1) records.'
コードの解説をします。
テストAccountを作ります。
その元になるパラメータを定義し、Accountのcreateで作成します。
test_AccountApi.py
account = {
'account_name' : "search_account",
'start_on' : '2021-05-23 00:00:00',
'end_on' : '2030-12-31 00:00:00',
'created_by' : 999,
'created_at' : datetime.datetime.now(),
'updated_by' : 999,
'updated_at' : datetime.datetime.now(),
'status' : Status.getStatusKey("NEW")
}
# createのテスト
assert Account.create(account, 999) == True
次に検索をして作成されたAccountのIDを調べます。
検索の条件パラメータを定義してAPIから実行します。
test_AccountApi.py
payload = {
"account_name":"search_account",
"start_on":"2021-05-23 00:00:00",
"end_on":"2030-12-31 00:00:00"
}
# APIから確認
url = f"http://localhost:5000/api/account/search"
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, json=payload)
レスポンスで返されたJSONの各項目を確認します。
test_AccountApi.py
# HTTP Statusコードが200であること
assert response.status_code == 200
print(f"test_account_search():json response.text={response.text}")
# BODYをjsonでパースできること
data = json.loads(response.text)
print(f"test_account_search():json data={data}")
assert data['body'][0]['account_name'] == payload["account_name"]
assert data['body'][0]['start_on'] == payload["start_on"]
assert data['body'][0]['end_on'] == payload["end_on"]
assert data['body'][0]['created_by'] == 999
assert data['status']['code'] == 'I0001'
assert data['status']['message'] == 'Found (1) records.'