APIを使って色々やる場面があるかと思います。
以下の記事でも緯度経度から標高を取得するAPIを使いました。
開発をしていく中でAPIを利用する側はある程度できていてもAPI側がまだ未実装という場合が割とあると思います。そうすると実装が終わってからでないと実際の動作確認ができないということになります。ですが、簡単なローカルAPIを作って定数でも返すようにすればAPI側がまだ実装されていなくてもある程度動作確認ができるので便利です。
調べてみるとpythonで割と簡単にローカルAPIサーバを作れることが判明したので、今回はその備忘録になります。
では、始めます。
０：前提
今回はAPIとして以下のものを考えていきます。
|
GET API |
UPDATE API |
| URL |
http://localhost:3000/ |
http://localhost:4000/ |
| Content-type |
json, utf-8 |
json, utf-8 |
| request（例） |
なし |
{
"data" : {
"name" : "テスト",
"value" : 100
}
} |
| response（例） |
{
"data" : {
"name" : "テスト",
"value" : 100
},
"call_count": 0
} |
{
"result" : "OK",
"data" : {
"name" : "テスト",
"value" : 100
}
} |
簡単に説明すると以下のような動きを想定したAPIになります。
- GET API、UDATE APIどちらもjson形式でやりとりする
- GET APIのcall_countはGET APIが呼ばれた回数の値とする
- UDATE APIではrequestのjsonの値でnameとvalueが更新される
- UDATE APIの戻り値のresultは更新できた場合OK、失敗した場合NGの文字列を設定する
かなりざっくりとしたものですが、とりあえず単純なGETとUPDATEのAPIを作っていくという感じです。
１：ローカルAPIサーバの作り方
書き方は簡単で以下の記述でjsonを返すローカルAPIサーバを作ることができます。
・sample_server.py
from wsgiref.simple_server import make_server
import json
API_PORT = 2000
def getStringByteSize(str):
return len(str.encode('utf-8'))
def app(environ, response):
json_data = {'message':'hoge'}
content_length = getStringByteSize(json.dumps(json_data))
status = '200 OK'
header = [
('Access-Control-Allow-Origin', '*'),
('Access-Control-Allow-Methods', '*'),
('Access-Control-Allow-Headers', "X-Requested-With, Origin, X-Csrftoken, Content-Type, Accept"),
('Content-type', 'application/json; charset=utf-8'),
('Content-Length', str(content_length))
]
response(status, header)
return [json.dumps(json_data).encode("utf-8")]
with make_server('', API_PORT, app) as httpd:
print("Start API. URL is http://localhost:" + str(API_PORT) + "/ ...")
httpd.serve_forever()
PowerShellなりコマンドプロンプトなりを起動させて上記のプログラムを実行すると、以下のようにURLが表示されてサーバが起動します。
$ python sample_server.py
Start API. URL is http://localhost:2000/ ...
この状態でブラウザを開いて「http://localhost:2000/」とURLを打ち込んで表示させると、以下のようにソースコードにおいてresponseとして定義したjsonの値が表示されます。
requestでjsonが来た場合は、コメントアウト部分に書いてある方法で取得することができます。
このサーバを止める場合は「Ctrl + C」キー、それでも止まらない場合は「Ctrl + Pause」キーで止めます。
ちなみにヘッダー部分ですが、これらについてあまり詳しく調べられていないので今回は詳しく言及しません。
簡単に言うとセキュリティ関係やデータ形式をどうするかという記述をしている部分なので、これらを記述していないとブラウザ側でエラーとなる場合があるので記述しておいてください。あくまでテスト用として許可しているだけなので、このヘッダー設定のまま実運用するとセキュリティ面で問題が発生するので注意してください。
２：ローカルAPIサーバサンプル
書き方がわかったので、実際に「０：前提」に書いた想定のローカルAPIサーバサンプルを以下に示します。またそれらがちゃんと動いているかを確認するためのhtmlとjavascriptも載せておきます（2021/05/29：ソースコードを少し修正しました）。
・local_test_api.py
from wsgiref.simple_server import make_server
import json
import threading
import time
GET_API_PORT = 3000
UPDATE_API_PORT = 4000
GET_API_JSON_PATH = "./get_api.json"
UPDATE_API_JSON_PATH = "./update_api.json"
STATUS = '200 OK'
call_count = 0
get_data_json = None
update_data_json = None
def main():
global get_data_json
get_data_json = load_json_file(GET_API_JSON_PATH)
global update_data_json
update_data_json = load_json_file(UPDATE_API_JSON_PATH)
get_app_thread = threading.Thread(target=start_get_api_server)
get_app_thread.setDaemon(True)
get_app_thread.start()
update_app_thread = threading.Thread(target=start_update_api_server)
update_app_thread.setDaemon(True)
update_app_thread.start()
while True:
time.sleep(1)
def load_json_file(json_path):
with open(json_path, "r", encoding="utf-8_sig") as f:
json_data = json.load(f)
return json_data
def start_get_api_server():
with make_server('', GET_API_PORT, get_app) as httpd:
print("Start Get API. URL is http://localhost:" + str(GET_API_PORT) + "/ ...")
httpd.serve_forever()
def get_app(environ, response):
global call_count
call_count = call_count + 1
get_data_json['call_count'] = call_count
content_length = getStringByteSize(json.dumps(get_data_json))
header = [
('Access-Control-Allow-Origin', '*'),
('Access-Control-Allow-Methods', '*'),
('Access-Control-Allow-Headers', "X-Requested-With, Origin, X-Csrftoken, Content-Type, Accept"),
('Content-type', 'application/json; charset=utf-8'),
('Content-Length', str(content_length))
]
response(STATUS, header)
return [json.dumps(get_data_json).encode("utf-8")]
def start_update_api_server():
with make_server('', UPDATE_API_PORT, update_app) as httpd:
print("Start Update API. URL is http://localhost:" + str(UPDATE_API_PORT) + "/ ...")
httpd.serve_forever()
def update_app(environ, response):
wsgi_input = environ["wsgi.input"]
try:
request_content_length = int(environ["CONTENT_LENGTH"])
request_data = json.loads(wsgi_input.read(request_content_length))
print(request_data)
if request_data['data']['name'] == "":
update_data_json['result'] = "NG"
else:
update_data_json['result'] = "OK"
update_data_json['data']['name'] = request_data['data']['name']
update_data_json['data']['value'] = int(request_data['data']['value'])
get_data_json['data']['name'] = request_data['data']['name']
get_data_json['data']['value'] = int(request_data['data']['value'])
except ValueError:
print("--- error ---")
content_length = getStringByteSize(json.dumps(update_data_json))
header = [
('Access-Control-Allow-Origin', '*'),
('Access-Control-Allow-Methods', '*'),
('Access-Control-Allow-Headers', "X-Requested-With, Origin, X-Csrftoken, Content-Type, Accept"),
('Content-type', 'application/json; charset=utf-8'),
('Content-Length', str(content_length))
]
response(STATUS, header)
return [json.dumps(update_data_json).encode("utf-8")]
def getStringByteSize(str):
return len(str.encode('utf-8'))
if __name__ == '__main__':
main()
・get_api.json
{
"data" : {
"name" : "テスト",
"value" : 100
},
"call_count": 0
}
・update_api.json
{
"result" : "OK",
"data" : {
"name" : "テスト",
"value" : 100
}
}
・index.html
<html lang="ja">
<head>
<meta charset = "utf-8">
<meta http-equiv="Cache-Control" content="no-cache">
<title>ローカルAPIサーバテスト</title>
<script src="https://code.jquery.com/jquery-3.6.0.js"></script>
</head>
<body>
name:<input id="name" value="">
<br>
value:<input id="value" value="">
<br>
<button id="update_button" onclick="update()">UPDATE</button>
<button id="get_button" onclick="get()">GET</button>
<br>
<div id="text">ここに結果が表示されます</div>
<script src="./script.js"></script>
</body>
</html>
・script.js
var update_connect_flg = false;
var get_connect_flg = false;
function getApi() {
return $.ajax({
url : 'http://localhost:3000/',
type : 'GET',
dataType : 'JSON',
scriptCharset : 'utf-8',
timespan : 100
});
}
function updateApi(send_json) {
var json_data = JSON.stringify(send_json);
return $.ajax({
url : 'http://localhost:4000/',
type : 'POST',
data : json_data,
dataType : 'JSON',
contentType : 'application/json; charset=utf-8',
contentLength : json_data.length,
timespan : 100
});
}
function get() {
if (get_connect_flg) { return; }
get_connect_flg = true;
getApi().done(function(res) {
console.log(res);
let res_data = "[GET response]<br>";
res_data += "name: " + res.data.name + "<br>";
res_data += "value: " + String(res.data.value) + "<br>";
res_data += "call_count: " + String(res.call_count);
$('#text').html(res_data);
}).fail(function() {
$('#text').html("GETに失敗しました");
}).always(function() {
get_connect_flg = false;
});
}
function update() {
if (update_connect_flg) { return; }
const name = $('#name').val();
const value = $('#value').val();
const send_json = {
'data' : {
'name' : name,
'value' : parseInt(value)
}
};
console.log(send_json);
update_connect_flg = true;
updateApi(send_json).done(function(res) {
console.log(res);
if ("OK" == res.result) {
let res_data = "[UPDATE response]<br>";
res_data += "result: " + res.result + "<br>";
res_data += "name: " + res.data.name + "<br>";
res_data += "value: " + String(res.data.value) + "<br>";
$('#text').html(res_data);
} else {
let res_data = "[UPDATE response]<br>";
res_data += "result: " + res.result + "<br>";
res_data += "name: " + res.data.name + "<br>";
res_data += "value: " + String(res.data.value) + "<br>";
$('#text').html("更新に失敗しました<br>" + res_data);
}
}).fail(function(res) {
$('#text').html("UPDATEに失敗しました");
}).always(function() {
update_connect_flg = false;
});
}
使い方として、まず「local_test_api.py」を実行してGetとUpdateのAPIサーバを起動させます。
$ python local_test_api.py
Start Update API. URL is http://localhost:4000/ ...
Start Get API. URL is http://localhost:3000/ ...
このサーバを起動させた状態で「index.html」を立ち上げて「GET」ボタンをクリックするとGetのAPIが、「UPDATE」ボタンをクリックするとinputタグに入力された値でUpdate APIが呼ばれ、結果がdivタグ部分に表示されます。
実際に動かすと以下のようになります。
少し説明すると「local_test_api.py」でGetとUpdateのAPIサーバをスレッドとして個別に実行しています。
APIサーバが動いている状態で「index.html」を立ち上げて「GET」ボタンをクリックするとjavascriptのget( )関数が実行され、GetのAPIの結果を取得してdivタグ部分に表示させています。「UPDATE」ボタンをクリックするとjavascriptのupdate( )関数が実行され、inputタグに入力された値で作ったjsonを使ってUpdate APIが呼ばれ、その結果をdivタグ部分に表示させています。
またNGを再現するために、nameの値が空の場合にNGを返すようにAPIサーバ側で処理をしています。
以上がpythonとjavascript、htmlで簡単なローカルAPIサーバを作る方法です。
本当はpythonだけで完結しても良かったのですが、画面に表示させたりした方がいいかと思いそちらも載せてみました。
その結果ソースコード部分がかなり長くなってしまいましたが、このサンプルを少し改造すれば割と応用が利くかと思います。
・2021/05/25:追記
今回のサンプルソースを一応公開しておきます。
・参考資料
