The fast API receives both websocket and HTTP requests. Websocket will delay HTTP requests. What should I do?

[lemonit-eric-mao]

python3.6

fastapi 0.63.0

server

#!/usr/bin/python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/25 11:28
# @Author  : Eric.Mao
# @FileName: main.py
# @Software: PyCharm
# @Blog    : http://www.dev-share.top/

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    allow_origins=["*"],
)

if __name__ == '__main__':
    uvicorn.run('main:app', host='localhost', port=8000, reload=True, debug=True, log_level="debug")

'''
========================================================================================
'''

import time
from fastapi import WebSocket, WebSocketDisconnect


# websocket test
@app.websocket("/websocket")
async def mapping_exec(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            receive_text = await websocket.receive_text()
            print(receive_text)
            time.sleep(30)
    except WebSocketDisconnect:
        print('客户端重新链接!')


# http test
@app.get("/mapping/{page_no}/{page_size}/{mapping_name}", tags=["mapping"])
def query_mapping_all(page_no: int, page_size: int, mapping_name: str):
    print('请求成功')

---

Implementation steps:

1. The client sends information to websocket and makes it wait for 30 seconds at the server

2. The client is sending the HTTP request. At this time, the HTTP request will be in pending, and it will not continue to process until the websocket wait ends for 30 seconds

I don't know why it's blocked. I'll ask experts to give me more advice

---

200	20

200	20

[Mause]

You cannot use time.sleep in async functions, it will indeed block as you have experienced - you should instead use await asyncio.sleep(30)

[lemonit-eric-mao]

await asyncio.sleep(30)

First of all, thank you for your answer. I have another question to ask. If I execute a function, it's just a common method and can't use async / await, what should I do?

I made the following changes

import time
import asyncio
from fastapi import WebSocket, WebSocketDisconnect


# websocket test
@app.websocket("/websocket")
async def mapping_exec(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            receive_text = await websocket.receive_text()
            # OK
            # await asyncio.sleep(30)
            # I made the following changes
            asyncio.get_event_loop().run_until_complete(test_sleep())
            print(receive_text)
    except WebSocketDisconnect:
        print('客户端重新链接!')


# I made the following changes
def test_sleep():
    time.sleep(30)

# http test
@app.get("/mapping/{page_no}/{page_size}/{mapping_name}", tags=["mapping"])
def query_mapping_all(page_no: int, page_size: int, mapping_name: str):
    print('请求成功')

Obviously, I can't do this. What should I do?

[ycd]

Just do asyncio.sleep().

async def mapping_exec(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            receive_text = await websocket.receive_text()
            await asyncio.sleep(30)
            print(receive_text)

[lemonit-eric-mao]

@ycd
Thank you for your reply. I want to use def test_ sleep (): it means that there will be a lot of computation in it, which takes a long time
This function means that I use a third-party tool for calculation, and its function does not support async/await

[ycd]

Then use def instead of async def when your endpoint does heavy computation.

By default FastAPI will use run_in_threadpool from Starlette and it uses loop.run_in_executor underneath.

run_in_executor will execute the function in the default loops executor also you might want to check options like ProcessPoolExecutor and ThreadPoolExecutor.

So to tidy up, use def instead.

[falkben]

@ycd Isn't a requirement of websockets async? Of course, I don't fully understand whether websockets is a requirement in the original question...

If the work is going to be heavy cpu, using ProcessPoolExecutor seems like a straightforward thing to do. Alternatively, you could run it as a subprocess: asyncio.create_subprocess_exec

[lemonit-eric-mao]

@falkben
Hello, expert. Thank you very much for your advice. I have made the following changes to the code. I don't know if I understand it correctly. It will still block @app.get("/mapping/{page_no}/{page_size}/{mapping_name}", tags=["mapping"]) Its request.

from fastapi import WebSocket, WebSocketDisconnect


# websocket test
@app.websocket("/websocket")
async def mapping_exec(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            receive_text = await websocket.receive_text()

            # import concurrent.futures
            # with concurrent.futures.ProcessPoolExecutor() as executor:
            #     # It will still block
            #     executor.submit(test_sleep).done()

            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor() as executor:
                # It will still block
                executor.submit(test_sleep).done()

            print(receive_text)
    except WebSocketDisconnect:
        print('客户端重新链接!')


# I made the following changes
def test_sleep():
    # time.sleep(30)
    while True:
        pass

# http test
@app.get("/mapping/{page_no}/{page_size}/{mapping_name}", tags=["mapping"])
def query_mapping_all(page_no: int, page_size: int, mapping_name: str):
    print('SUCCESS')

[lemonit-eric-mao]

This is how I deal with it. I changed the way I use it and let websocket initiate communication from the server. Thank you very much for your answers

Three party tool callback method class

#!/usr/bin/python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/24 14:28
# @Author  : Eric.Mao
# @FileName: result_callback.py
# @Software: PyCharm
# @Blog    : http://www.dev-share.top/

import asyncio
# ellipsis .....................
from ansible.plugins.callback import CallbackBase


class ResultCallback(CallbackBase):

    def __init__(self, websocket, *args, **kwargs):
        super(ResultCallback, self).__init__(*args, **kwargs)
        self.websocket = websocket

    # 
    def v2_runner_on_ok(self, result):
        result_json = {
            'stats': 'ok',
            'task': result._task.name,
            'hostname': result._host.name,
            'logs': result._result,
        }
         # OK
        asyncio.new_event_loop().run_until_complete(self.websocket.send_json(result_json))

---

#!/usr/bin/python
# -*- coding: utf-8 -*-
# @Time    : 2020/11/23 16:28
# @Author  : Eric.Mao
# @FileName: ansible_playbook.py
# @Software: PyCharm
# @Blog    : http://www.dev-share.top/

import asyncio
# ellipsis .....................
from opt_ansible.result_callback import ResultCallback


class AnsiblePlaybook(object):

    def __init__(self):
        pass
    def run(self, websocket):
    # ellipsis .....................
        try:
            playbook = PlaybookExecutor(playbooks=[self.script_tmp_path],
                                        inventory=inventory,
                                        variable_manager=variable_manager,
                                        loader=loader,
                                        passwords=self.password)
            # Use tripartite class
            playbook._tqm._stdout_callback = ResultCallback(websocket)
            playbook.run()
        except Exception as e:
            print(e.message)
            result_json = {
                'stats': 'exception',
                'message': e.message
            }
            asyncio.new_event_loop().run_until_complete(websocket.send_json(result_json))
        finally:
            os.remove(self.inventory_tmp_path)
            os.remove(self.script_tmp_path)
            shutil.rmtree(constants.DEFAULT_LOCAL_TMP, True)

[tiangolo]

Thanks for the help here everyone! 👏 🙇

Thanks for reporting back and closing the issue @lemonit-eric-mao 👍

Sorry for the long delay! 🙈 I wanted to personally address each issue/PR and they piled up through time, but now I'm checking each one in order.