0

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

0

この質問に回答するには、ログインする必要があります。

のタグが付いた他の質問を参照する。