FastAPI pre queue

[willnufe]

I have built a service using FastAPI and FastStream. Given that the peak QPS of requests exceeds the service's processing capacity, I need a queue to cache messages so that consumers can retrieve and process them sequentially.

Could you please review my code and assess whether the designed structure is reasonable? I've noticed that while publishing is fast, it takes a long time for the client to receive synchronous responses.

from faststream.rabbit.fastapi import RabbitRouter
from pydantic import BaseModel



import os
import uuid
import time


import uvicorn
from fastapi import FastAPI, Request, BackgroundTasks
from contextlib import asynccontextmanager
import aiohttp

from download_file_async import Audio_file_download
from vad_process import Silero_Vad_Api



FILE_ENDPOINT = os.getenv("FILE_ENDPOINT")
TOKEN_ENDPOINT = os.getenv("TOKEN_ENDPOINT")



class Message1(BaseModel):
    session_id: str
    file_id: str
    audio_type: str
    url: str
    customData: dict



class AppContext:
    def __init__(self):

        self.asyncEngine = None
        self.download = None 


@asynccontextmanager
async def lifespan(app: FastAPI):

    app_ctx = AppContext()

    connector_d = aiohttp.TCPConnector(limit=20, verify_ssl=False)
    session_d = aiohttp.ClientSession(connector=connector_d)
    app_ctx.download = Audio_file_download(TOKEN_ENDPOINT, FILE_ENDPOINT, session_d)

    app_ctx.asyncEngine = Silero_Vad_Api()

    app.state.app_ctx = app_ctx


    yield


    app.state.app_ctx = None



rabbit_url = 'amqp://guest:guest@127.0.0.1/'
router = RabbitRouter(rabbit_url, lifespan=lifespan, max_consumers=4, log_level=3)
app = FastAPI()



@router.post("/test/")
async def generate(raw_request: Request):

    try:

        request_dict = await raw_request.json()

        session_id = request_dict['session']['session_id']
        sync_return = {
            "result":{
                "session_id": session_id,
                "status": 200,
                "message": "success"
            }
        }


        file_id = request_dict['source']['file_id']
        audio_type = request_dict['source']['audio_type']
        url = request_dict['response_route']['url']

        response_dict = {
            'session_id': session_id,
            'file_id': file_id,
            'audio_type': audio_type,
            'url': url,
            'customData': request_dict['customData']
        }
        await router.broker.publish(Message1(**response_dict), "pre_queue")


        return sync_return

    except Exception as e:
        log.logger.error(f"[stepPre: rabbit]: {e}")
        sync_return['result']['status'] = 201
        sync_return['result']['message'] = f"{e}"
        return sync_return  




@router.subscriber("pre_queue")
async def message_consumer(m: Message1):

    try:

        app_ctx = app.state.app_ctx

        session_id = m.session_id
        file_id = m.file_id
        audio_type = m.audio_type
        url = m.url
        customData = m.customData


		########
		[processing...]
		########

            
    except Exception as e:
        log.logger.error(f"[messageConsumerInner]: {e}")






app.include_router(router)

if __name__ == "__main__":

    uvicorn.run(
        "__main__:app",
        host='0.0.0.0',
        port=8080,
        workers=4,
        limit_concurrency=12,
        reload=False
    )