本記事は
flaskを使ってRestAPIサーバを作ってみるの「参照」について記述しています。
ここまでの流れはflaskを使ってRestAPIサーバを作ってみるを参照してください。
参照編で説明するコードは feature/chapter2 ブランチにあります。
flask_sv プロジェクトのディレクトリ配下で次のコマンドを実行してください。
$ git checkout feature/chapter2
1. 参照
1.1. モデル層・エンティティ
accountをaccount_idからレコードを抽出してAPIで返却するところまでを説明します。
importは省略するのでgithubのコードを参照してください。
まずモデル層の核になるAccountクラスです。
# model class
class Account(Base):
"""
accountモデル
flask_svシステムにログインするアカウントを管理するモデル
Parameters
----------
Base : データベース接続子
"""
__tablename__ = 'account'
id = Column(Integer, primary_key=True)
account_name = Column(String())
start_on = Column(Timestamp)
end_on = Column(Timestamp)
created_by = Column(Integer)
created_at = Column(Timestamp)
updated_by = Column(Integer)
updated_at = Column(Timestamp)
status = Column(Integer)
# get Dict data
def toDict(self):
"""
ディクショナリ形式でクラスの全メンバを返却する
Parameters
----------
self : 自クラス
Returns
-------
クラスの全メンバのディクショナリ
"""
return {
'id' : int(self.id),
'account_name' : str(self.account_name),
'start_on' : str(self.start_on),
'end_on' : str(self.end_on),
'created_by' : int(self.created_by),
'created_at' : str(self.created_at),
'updated_by' : int(self.updated_by),
'updated_at' : str(self.updated_at),
'status' : Status.getStatusName(str(self.status))
}
...
__tablename__にテーブル名。
Accountクラスにカラムのデータ構造を定義します。
オブジェクトのメンバをディクショナリで返せるtoDict関数を定義しました。
このコードの中にそれぞれ、参照、作成、検索、更新、削除の関数を作っていきます。
参照のモデル層の実装は次のコードです。
def getById(account_id, operation_account_id):
"""
アカウントidでaccountテーブルを検索をし、該当したAccountオブジェクト群を取得する
Parameters
----------
account_id : 検索対象のアカウントid
operation_account_id : 操作ユーザーのアカウントid
Returns
-------
Accountオブジェクトのリスト
"""
Session = sessionmaker(bind=engine)
ses = Session()
res = ses.query(Account).get(account_id)
ses.close()
return res
コードの解説をします。
account_id(検索対象のアカウントid)とoperation_account_id(操作ユーザーのアカウントid)を受け取りDBとのセッションを開きます。
Session = sessionmaker(bind=engine)
ses = Session()
次の行でSQLを実行します。
getはwhere句で主キーを指定することができます。
res = ses.query(Account).get(account_id)
次の行はセッションを閉じます。
ses.close()
SQLの結果を返却します。
SQLのselect文で取得した結果はそのデータ型(ここではAccount型)のリストですが、getで取得した場合は１レコードが特定できるので単一データを返します。
return res
本当は色々漏れがあり、Exceptionが起きたときの対処など必要ですが、後で改善していきます。
課題 1.1
上記、Exceptionが起きたときにAPIを利用している人にエラーの内容を伝えられる様APIを改修し /account/get_with_message を作成してください。
1.2. ドメイン層・ビジネスロジック
このレイヤーでロジックを実装することにします。ロジックは業務に紐づいたコードを書き、モデル層、アプリケーション層にデータにまつわる機能を寄せましょう。
Accountを呼び出すのはrestapiパッケージのAccountApi.pyです。
Account.pyではDBとの入出力に特化しました。
このコードは内部のデータ構造(pythonのDICTデータ)からWeb APIのJSONデータに変換する機能を持たせます。
ドメイン層の実装は次のコードです。
def getById(account_id, operation_account_id):
"""
/account/get/<id>で呼び出されたAPIの検索処理
Parameters
----------
account_id : int
検索するアカウントのアカウントID
operation_account_id : int
Webアプリケーション操作アカウントのID
Returns
-------
ret
json形式のアカウント詳細
{
"body": {
"name": "account",
"id": <account_id>,
"account_name": <account_name>,
"start_on": "2021-01-01 10:00:00",
"end_on": "2025-12-31 21:00:00"
},
"status": {
"code" : "I0001",
"message" : "",
"detail" : ""
}
}
"""
result = Account.getById(account_id, operation_account_id)
# TODO モデルの検索結果(正常・異常)によってレスポンスの出力内容を変える
result_json = {
"body": {
"name": "account",
"id": account_id,
"account_name": result.account_name,
"start_on": result.start_on.strftime("%Y-%m-%d %H:%M:%S"),
"end_on": result.end_on.strftime("%Y-%m-%d %H:%M:%S")
},
"status": {
"code" : "I0001",
"message" : "",
"detail" : ""
}
}
return result_json
コードの解説をします。
次の行でmodel層のAccountの関数を実行しています。
result = Account.getById(account_id, operation_account_id)
JSONに変換するため、時間も文字列で表現しなければならないので文字列の値を持つDICTに変換しています。
# TODO モデルの検索結果(正常・異常)によってレスポンスの出力内容を変える
result_json = {
"body": {
"name": "account",
"id": account_id,
"account_name": result.account_name,
"start_on": result.start_on.strftime("%Y-%m-%d %H:%M:%S"),
"end_on": result.end_on.strftime("%Y-%m-%d %H:%M:%S")
},
"status": {
"code" : "I0001",
"message" : "",
"detail" : ""
}
}
変換した文字列だけのDICTをJSONデータに変更するのは。アプリケーション層に任せることにします。
return result_json
課題 1.2
現在の実装では、正常時しか動きません。Accountモデルの検索結果(正常・異常)によってレスポンスの出力内容を変え、指定したアカウントIDが存在しない時はワーニングとして、status.codeにW0002を設定。エラー時は、status.codeにE000３。必要があればstatus.messageに必要なメッセージを入れて返却する様にしましょう。
1.3. アプリケーション層・フロントへのIF
このコードでサーバーで受け取ったURLと呼び出す関数とを紐付けしています。HTTPのメソッドの判定もこのコードで実装しています。
これはGETメソッドで http://サーバーのIP/account/get/100 にアクセスしたらaccount_id=100でAccountApi.getByIdを呼び出します。
処理の結果はテキストでDICTに埋め込むことでJSON形式にフォーマットにしていますが、jsonifyはエンコード、改行コード、Responseヘッダの追加などをします。
@api_bp.route('/account/get/<id>', methods=['GET'])
def getAccount(id):
account_json = AccountApi.getById(id, system_account_id)
return jsonify(account_json)
ここではエンコードや区切り文字を置き換えます。
jsonを返すエンコーダーについて調べるとjson.dumpsを使う例もよく見つかります。違いはflask jsonify vs json.dumps 違いを簡単に解説を参照してください。
1.4. テストコードからの実行
テストコード全文
def test_account_get():
"""
restapi/getById
"""
account = {
'account_name' : 'flask_sv',
'start_on' : '2021-05-05 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")
}
Account.create(account, 999) == True
account_dict = {
'account_name' : "flask_sv",
'end_on' : "2030-12-31 00:00:00"
}
result = Account.search(account_dict, 1)
account_id = result[0].id
# APIから確認
url = f"http://localhost:5000/api/account/get/{account_id}"
headers = {'Accept-Encoding': 'identity, deflate, compress, gzip',
'Accept': '*/*', 'User-Agent': 'flask_sv/0.0.1',
'Content-type': 'application/json; charset=utf-8',
}
response = requests.get(url, headers=headers)
assert response.status_code == 200
data = json.loads(response.text)
assert data['body']['name'] == "account"
assert data['body']['account_name'] == "flask_sv"
assert data['body']['start_on'] == "2021-05-05 00:00:00"
assert data['body']['end_on'] == "2030-12-31 00:00:00"
assert data['status']['code'] == "I0001"
コードの解説をします。
テストAccountを作ります。その元になるパラメータです。
account = {
'account_name' : 'flask_sv',
'start_on' : '2021-05-05 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")
}
上記のパラメータをAccountのcreateで作成します。
createはまだ出てきていませんが2.作成で説明します。
Account.create(account, 999) == True
次に検索をして作成されたAccountのIDを調べます。
検索もまだ出てきていませんが3.検索で説明します。
account_dict = {
'account_name' : "flask_sv",
'end_on' : "2030-12-31 00:00:00"
}
result = Account.search(account_dict, 1)
result_id = result[0].id
テスト対象のAPIを実行します。
# APIから確認
url = f"http://localhost:5000/api/account/get/{account_id}"
headers = {'Accept-Encoding': 'identity, deflate, compress, gzip',
'Accept': '*/*', 'User-Agent': 'flask_sv/0.0.1',
'Content-type': 'application/json; charset=utf-8',
}
response = requests.get(url, headers=headers)
返却されたレスポンスの結果を確認します。
assert response.status_code == 200
data = json.loads(response.text)
assert data['body']['name'] == "account"
assert data['body']['account_name'] == "flask_sv"
assert data['body']['start_on'] == "2021-05-05 00:00:00"
assert data['body']['end_on'] == "2030-12-31 00:00:00"
assert data['status']['code'] == "I0001"
assert data['status']['message'] == ""
テストプログラムは次の様に実行します。
pytest -v -s tests/domain/test_AccountApi.py
============== test session starts ===========
...
... ::test_account_get PASSED
...
pytestは、pythonのテストライブラリです。
flask_svではpipenv installを実行した時にインストールされます。
実行後、PASSEDと表示されると想定通りの結果が出て試験の条件を満たしたことが確認できます。