

FastApiとUvicornを使用したAPIサーバープロセスとアプリケーションのMainプロセスとの通信について悩んでおります。

# アプリケーション内の構成イメージ

[![プロセスイメージ](https://i.stack.imgur.com/5qLrj.gif)](https://i.stack.imgur.com/5qLrj.gif)

# 動作説明

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接続を利用  
調べたところ分かりませんでした。

※最悪の対処方法としては、思いつく方法はあります。  
１）POSTメソッドで、DBテーブルにPOST内容を書き込む。  
２）MainProcessが定期的にDBテーブルをチェックし、要求事項があったら処理する。  
３）MainProcessが結果テーブルに処理結果を書き込む。  
４）PostMethodが結果テーブルを参照して返信する。  
これはかなりパフォーマンスが悪いことは容易に想像がつきます。

# ソースコード

実行後に、WEBブラウザで、http://127.0.0.1:8008/docs にアクセスするとAPIテストページが表示されます。  
ソースコードが多くなり申し訳ありません。最低限動作する程度に絞ったつもりです。図中で必要のないProcessやエラー処理などは省略してあります。本来DELETEメソッドや多くのPOSTメソッド(用途によってソースファイルを分けている)があります。

[code]

    フォルダ/ファイル構成
    (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(実行スクリプト)
    
    
[/code]

app.py

[code]

    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()
    
    
[/code]

api_root.py

[code]

    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()
    
[/code]

api_post_work.py

[code]

    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)
    
[/code]

api_server.py

[code]

    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
    
[/code]

com_clinet.py

[code]

    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
    
[/code]

com_server.py

[code]

    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
    
[/code]


CC BY-SA 4.0
2022-04-22T02:59:51.463
88446
2022-04-22T02:59:51.463
32891
question
python3 python-multiprocessing fastapi
FastAPI+Uvicorn マルチプロセス実行環境でメインプロセスと通信する方法