はじめに
こんにちは、AIエンジニアの山下です。
これまではweb hookを利用したAPIの解析結果を確認・テストするのにPostmanとwebhook.siteを使用していました。
少し確認する程度ならこれでもよかったのですが、何度もテストを行いたい場合などは毎回ブラウザーを開いて確認するとなると結構面倒。
そこで、python＋Flask（Webアプリケーションフレームワーク）で簡易的なローカルサーバーを構築し、ngrokというツール（ローカルサーバーと外部ネットワークを繋ぐツール）を使用し、ブラウザーやターミナルを開かず、jupyter notebook内でAPIへのリクエストの送信からローカルサーバーでのレスポンスのデータの保存までできることを目指しました。
(＊ネットワークやwebアプリケーションに慣れていないので細かい部分で認識が間違っていたらすみませんmm)
＊OSはMac。
前提として
- ngrokにユーザー登録が済んでいる
- 今回想定しているのはwebAPIにリクエストを送り、そのレスポンス（解析結果）をweb hookとして送る（指定したURLにPOSTリクエストする）という構成
- jupyter notebookでpythonモジュール（app.py）を読み込んだ上でリクエスト送信→APIからwebhookをローカルサーバーに送る→ローカルサーバーでレスポンス内容をjsonに変換したものを保存するまでの処理を行いたい
- ngrokにはngrokコマンドをインストールして使えるが、処理をjupyter notebook内で完結させたいため、Pythonラッパーのpyngrokというライブラリーを使用
- python内で”pip install flask pyengrok”が事前に必要
流れ
- 作成したpythonファイル（モジュール）とその説明
- 1.を読み込むjupyter notebookの説明
app.py
- 作成したモジュールです、この後に内容を説明していきます。
import json import logging import os import threading from logging import config from flask import Flask, jsonify, request from pyngrok import ngrok from werkzeug.serving import run_simple from config.logging_config import LOGGING_CONFIG class CreateApp(): ana_js = None def __init__(self, on_colab: bool = False) -> None: os.environ["FLASK_DEBUG"] = "development" # Open a ngrok tunnel to the HTTP server port = 8888 # ポート番号 public_url = ngrok.connect(port).public_url self.public_url = public_url.replace('http', 'https') # httpsに変換 print(" * ngrok tunnel \"{}\" -> \"http://127.0.0.1:{}\"".format( self.public_url, port)) app = Flask(__name__) app.debug = True app.config["SECRET_KEY"] = "secret key" # 要変更 # Update any base URLs to use the public ngrok URL app.config["BASE_URL"] = self.public_url app.config["ENV"] = "development" app.config['JSON_AS_ASCII'] = False # JSON日本語文字化け対策 # .. Update inbound traffic via APIs to use the public-facing ngrok URL # logging設定 config.dictConfig(LOGGING_CONFIG) logger = logging.getLogger() # Start the Flask server in a new thread if on_colab is False: threading.Thread( target=lambda: run_simple('localhost', port, app)).start() else: # on Colab, you can use the following code instead of the above # line threading.Thread( target=app.run, kwargs={"use_reloader": False}).start() # Define Flask routes @app.route("/", methods=['GET']) def index(): return "Web App with Python Flask!" @app.route('/receiver', methods=['POST']) def get_json(): js = request.get_json() if js is not None: # JSONが送られてきた場合、DB & ローカルに保存 logger.error('ERROR') self.ana_js = js # class変数に保存 with open(f'./result/{js["metadata"]["pdf_name"]}.json', 'w') as f: json.dump(js, f, indent=4) return jsonify(js) def receive(self) -> dict: while True: if self.ana_js is not None: return self.ana_js else: continue
pyngrokを使用し、HTTPサーバーを開く
- port は何でも良さそうでdefaultだと5000になります
- public_urlがngrokが作成したトンネルのURLになります
# Open a ngrok tunnel to the HTTP server port = 8888 # ポート番号 public_url = ngrok.connect(port).public_url self.public_url = public_url.replace('http', 'https') # httpsに変換 print(" * ngrok tunnel \"{}\" -> \"http://127.0.0.1:{}\"".format( self.public_url, port))
Flask側の設定
- app.config[”BASE_URL”]に上記の”public_url”を指定する
- loggingも設定しましたが、本筋ではないので詳細は省きます
app = Flask(__name__) app.debug = True app.config["SECRET_KEY"] = "secret key" # 要変更 # Update any base URLs to use the public ngrok URL app.config["BASE_URL"] = self.public_url app.config["ENV"] = "development" app.config['JSON_AS_ASCII'] = False # JSON日本語文字化け対策 # .. Update inbound traffic via APIs to use the public-facing ngrok URL # logging設定 config.dictConfig(LOGGING_CONFIG) logger = logging.getLogger()
python HTTPの設定
- このあたりちゃんと理解できていませんが二つの処理を同時に実行させるために並列処理を行いFlaskサーバーをスレッドを使用し、実行しているようです
# Start the Flask server in a new thread if on_colab is False: threading.Thread( target=lambda: run_simple('localhost', port, app)).start()
- google Colab上だと以下
else: # on Colab, you can use the following code instead of the above # line threading.Thread( target=app.run, kwargs={"use_reloader": False}).start()
Flaskの処理とルーティング
# Define Flask routes @app.route("/", methods=['GET']) def index(): return "Web App with Python Flask!"
- こちら、ローカルサーバーで受け取ったレスポンスをjsonに変換し、ローカルの指定したディレクトリに保存する処理になります。（また、そのjsonをクラス変数に代入する処理も行ってます、この変数に代入されたデータは後ほど使います）
@app.route('/receiver', methods=['POST']) def get_json(): js = request.get_json() if js is not None: # JSONが送られてきた場合、DB & ローカルに保存 logger.error('ERROR') self.ana_js = js # class変数に保存 with open(f'./result/{js["metadata"]["pdf_name"]}.json', 'w') as f: json.dump(js, f, indent=4) return jsonify(js)
クラス内関数を定義
- APIにリクエストを送り、レスポンスが返ってくるまでにタイムラグがあり、ローカルサーバーにレスポンスが送られるまで実行し続ける関数になります
- 何のための関数か分かりづらいかもしれませんが、後のjupyter notebook内の全ての処理（セル）を一度の実行で行いため実装しています
def receive(self) -> dict: while True: if self.ana_js is not None: return self.ana_js else: continue
JupyterNotebook
- 1番目のセル
- 必要なライブラリを読み込んでます(app→先に作成したモジュール）
2番目のセル
- 1行目：ローカルサーバーの起動
- 2行目：public_urlの取得（パブリックngrokURL）
- 3行目：アクティブなトンネルをリストで取得
3番目のセル
＊url, api_key, pdf_path, dataの内容はダミー
4番目のセル
- app.pyのCreateAppクラス内関数receiveを実行
- このセルが実行完了（ローカルサーバーがレスポンスを受信、かつ、レスポンスjsonの保存）次第、次のセルが実行される
＊この処理を入れずに次のセルを実行するとローカルサーバーがレスポンスを受信する前にトンネルを閉じてしまう場合がある
5番目のセル
- 開いているトンネル閉じる
＊jupyter notebookを再起動するだけでトンネルは閉じられる
jupyter notebook コード
import json import requests from pyngrok import ngrok from app import CreateApp
# local server を起動
ca = CreateApp()
public_url = ca.public_url
tunnels = ngrok.get_tunnels()
# APIにデータを送信 url = 'https://OOOO.com/OOO' # ダミー api_key = 'api_key'. # ダミー webhook_url = public_url + '/receiver' pdf_path = 'OOOO/OOOO/OOO.pdf' metadata = {'pdf_name': '0000.pdf'} # postする data = { 'data:': 'data', webhook_url: 'webhook_url', metadata: {'hoge': 'test'}} js_str = json.dumps(data).encode('utf-8') response_local = requests.post( url, data=js_str, headers={'x-api-key': api_key}) print(f'response_local: {response_local}')
ca.receive().keys()
print(f'Before tunnels:\n{ngrok.get_tunnels()}') # Close all open ngrok tunnels _ = [ngrok.disconnect(tnnl.public_url) for tnnl in tunnels] print(f'After tunnels:\n{ngrok.get_tunnels()}')
最後に
webやネットワークに慣れてないこともあり、言語化するのに苦労しました。
（＊今回のものはあくまで、開発テスト確認用になります）
次はFlaskのローカルサーバーにログイン機能を実装してみたいと思います。
株式会社フォトラクションでは一緒に働く仲間を募集しています
