FastApiとUvicornを使用したAPIサーバープロセスとアプリケーションのMainプロセスとの通信について悩んでおります。
アプリケーション内の構成イメージ
動作説明
Uvicornを起動するServerProcess(の中のUvicorn動作Thred)とMain Process間のみでPOSTメソッド要求時に逐次利用するComServerを通じてやり取りを行おうと思いつきました。
これは、QueueによってUvicornプロセス内と通信する方法が分からなかったためです。
POSTメソッドで要求された内容をMainProcess内で各種情報を判断し、データベースへの反映やほかプロセスへの実行要求を行い、結果を生成して戻します。PostMethod処理内ではこの返信があるまで待ち、内容によってClientに返信します。※実際には処理に時間がかかるものは処理中として返信。応答がない場合にタイムアウトとして返信。
問題点
ComServer方式では、PostMethod側でKeep-aliveでコネクションを維持できず、頻度が高まるとコネクションリソースを食いつぶして、ComServerへの接続エラーを頻発してしまいます。
問い合わせ内容
どのような方法でも良いので、POSTメソッド要求時にMainProcessよって処理を実行し、その結果を返信する方法で良い方法はありますか?
1.PostMethodでアプリケーション動作期間中Keep-aliveでコネクションを維持する方法がありますか?
2.Uvicorn内と通信する方法が他にありますか?
・Queue利用
・ServerProcess内でのUvicornへのコールバック関数利用
・MainProcessとUvicorn内とWebSocket接続を利用
調べたところ分かりませんでした。
※最悪の対処方法としては、思いつく方法はあります。
1)POSTメソッドで、DBテーブルにPOST内容を書き込む。
2)MainProcessが定期的にDBテーブルをチェックし、要求事項があったら処理する。
3)MainProcessが結果テーブルに処理結果を書き込む。
4)PostMethodが結果テーブルを参照して返信する。
これはかなりパフォーマンスが悪いことは容易に想像がつきます。
ソースコード
実行後に、WEBブラウザで、http://127.0.0.1:8008/docs にアクセスするとAPIテストページが表示されます。
ソースコードが多くなり申し訳ありません。最低限動作する程度に絞ったつもりです。図中で必要のないProcessやエラー処理などは省略してあります。本来DELETEメソッドや多くのPOSTメソッド(用途によってソースファイルを分けている)があります。
フォルダ/ファイル構成
(root)
+-source
+-api
| +-post
| | +-api_post_work.py
| |
| +-api_root.py
|
+-api_server
| +-api_server.py
|
+-com
| +-com_client.py
| +-com_server.py
|
+-app.py(実行スクリプト)
app.py
from multiprocessing import freeze_support, set_start_method, Process
from multiprocessing import Queue as MP_Queue
import json
import os
import sys
import time
current_path = os.getcwd() + "/source"
path = current_path + "/api_server"
sys.path.append(path)
import api_server as API # nopep8
path = current_path + "/com"
sys.path.append(path)
import com_server as com # nopep8
class AppServer():
def run(self):
self.work_dict = dict()
self.com_send_queue, self.com_recv_queue = MP_Queue(), MP_Queue()
instCom = com.com_server()
instComProcess = Process(
name='ComServer',
target=instCom.run,
args=(self.com_send_queue, self.com_recv_queue)
)
instComProcess.start()
instAPI = API.ApiServer()
instAPIProcess = Process(
target=instAPI.run,
)
instAPIProcess.start()
while(True):
time.sleep(0.05)
self.get_Queue()
def get_Queue(self):
while(not self.com_recv_queue.empty()):
item = self.com_recv_queue.get()
if item:
ret_dict = dict()
msg = item.decode()
msg_dict = json.loads(msg)
if "type" in msg_dict:
msg_type = msg_dict["type"]
if msg_type == "GET":
if "work_name" in msg_dict:
work_name = msg_dict["work_name"]
if work_name in self.work_dict:
status = "SUCCESS"
result = self.work_dict[work_name]
ret_dict = {
"status": status,
"result": result,
"error": {}
}
elif msg_type == "POST":
if "work" in msg_dict:
work = msg_dict["work"]
if all(
[
"name" in work,
"category" in work,
]
):
name = work["name"]
category = work["category"]
if name in self.work_dict:
self.work_dict[name] = category
else:
self.work_dict.setdefault(name, category)
ret_dict = {
"status": "SUCCESS",
"result": {},
"error": {}
}
self.com_send_queue.put(json.dumps(ret_dict))
if __name__ == '__main__':
freeze_support()
set_start_method('spawn')
AppServer().run()
api_root.py
from fastapi import FastAPI
import os
import sys
current_path = os.getcwd() + "/source"
path = current_path + "/api"
sys.path.append(path)
from api_def import * # nopep8
path = current_path + "/api/post"
sys.path.append(path)
import api_post_work as post_work # nopep8
app = FastAPI(
title="Test",
description="Test",
version="0.1"
)
app.include_router(post_work.router)
@app.get("/")
def read_api_root():
result = ApiResult(
status="SUCCESS",
result={"Message": "Welcome!!"},
error={}
)
return result.dict()
api_post_work.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
import json
import os
import sys
current_path = os.getcwd() + "/source"
path = current_path + "/com"
sys.path.append(path)
import com_client as com # nopep8
class ApiWork(BaseModel):
category: str
router = APIRouter(
prefix=f"/work",
tags=["work"],
)
@router.post("/{work_name}")
def post_api_work(work_name: str, req_item: ApiWork):
msg_dict = {
"type": "POST",
"work": {
"name": work_name,
"category": req_item.category
}
}
recv_data = com.send_message(json.dumps(msg_dict))
if recv_data:
s_result = recv_data.decode()
if isinstance(s_result, str):
result = json.loads(s_result)
if 'status' in result:
if result['status'] == "ERROR":
raise HTTPException(status_code=400, detail=result)
if result:
return result
ret_dict = {
"status": "ERROR",
"result": {},
"error": {
"reason": "NOT RESPONSE"
}
}
raise HTTPException(status_code=400, detail=ret_dict)
api_server.py
from threading import Thread
import time
from uvicorn import Config, Server
import os
import sys
current_path = os.getcwd() + "/source"
path = current_path + "/api"
sys.path.append(path)
import api_root # nopep8
class web_server_thread:
def __init__(
self, web_app, host: str, port: int, reload: bool = False):
self.config = Config(
app=web_app,
host=host,
port=port,
reload=reload,
# log_level="warning",
)
self.server = Server(self.config)
self.server.install_signal_handlers = lambda: None
self.proc = None
def start(self):
self.th = Thread(name="WebServer Thread", target=self.server.run)
self.th.start()
def stop(self):
if self.th:
self.th.join(0.25)
class ApiServer():
def run(self, host: str = '127.0.0.1', port: int = 8008):
self.instThread = web_server_thread(
api_root.app, host=host, port=port)
self.instThread.start()
isLoop = True
while(isLoop):
time.sleep(0.2)
return
com_clinet.py
import asyncio
srv_addr = '127.0.0.1'
srv_port = 5638
async def client_message(message):
reader, writer = await asyncio.open_connection(srv_addr, srv_port)
writer.write(message.encode())
await writer.drain()
data = await reader.read()
writer.close()
await writer.wait_closed()
return data
def send_message(message):
recv_data = asyncio.run(client_message(message))
return recv_data
com_server.py
import asyncio
from queue import Queue
from threading import Thread
import time
srv_addr = '127.0.0.1'
srv_port = 5638
class com_server_protcol(asyncio.Protocol):
def __init__(self, send_queue):
self.transport = None
self.send_queue = send_queue
super().__init__()
return
def connection_made(self, transport):
self.transport = transport
client_address, client_port = self.transport.get_extra_info('peername')
# ホストマシン以外の接続は拒否
if client_address != srv_addr:
self.transport.close()
def data_received(self, data):
self.send_queue.put(data)
def connection_lost(self, exc):
self.transport.close()
class com_server():
def com_thread(self):
timer = None
time_out = 3.0
while(self.is_ThrLoop):
time.sleep(0.03)
while(not self.con_recv_queue.empty()):
item = self.con_recv_queue.get()
self.send_queue.put(item)
timer = time.time()
while(not self.recv_queue.empty()):
item = self.recv_queue.get()
if timer is not None:
self.instProt.transport.write(item.encode())
self.instProt.transport.close()
timer = None
# メッセージを受信してtime_out秒後に強制切断
if timer is not None:
if time.time() - timer > time_out:
self.instProt.transport.close()
timer = None
return
def run(self, recv_queue, send_queue):
self.is_ThrLoop = True
self.recv_queue, self.send_queue = recv_queue, send_queue
self.con_recv_queue = Queue()
self.Thr = Thread(name='com_server_thread', target=self.com_thread)
self.Thr.start()
self.event_loop = asyncio.get_event_loop()
self.instProt = com_server_protcol(self.con_recv_queue)
factory = self.event_loop.create_server(
lambda: self.instProt, srv_addr, srv_port)
self.server = self.event_loop.run_until_complete(factory)
try:
self.event_loop.run_forever()
finally:
self.is_ThrLoop = False
self.server.close()
self.event_loop.run_until_complete(self.server.wait_closed())
self.event_loop.close()
return