Kludex
commented
5 months ago Closed carlos-rian-quartile closed 5 months ago
Kludex
commented
5 months ago 
carlos-rian-quartile
commented
5 months ago @Kludex
I tested using AnyIO, and it worked correctly. I appreciate your help.
import anyio
import logging
import httpx
from fastapi import FastAPI, Query, WebSocket, WebSocketDisconnect
app = FastAPI()
logging.basicConfig(level=logging.DEBUG)
@app.websocket("/ws")
async def my_websocket(websocket: WebSocket, id: int = Query(...), name: str = Query(...)):
try:
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
except WebSocketDisconnect:
async with anyio.create_task_group() as _a:
with anyio.CancelScope(shield=True) as _b:
client = httpx.AsyncClient(timeout=10):
resp = await client.get("https://httpstat.us/200?sleep=5000")
logging.warning(f"websocket disconnected - id: {id} - {resp}")
logging.warning(f"websocket disconnected - id: {id}") # working
from fastapi.testclient import TestClient
def test_websocket():
client = TestClient(app)
with client.websocket_connect("/ws?id=1&name=John") as websocket:
data = "some data"
websocket.send_text(data)
msg = websocket.receive_text()
assert msg == f"Message text was: {data}"
test_websocket()
Initial Checks
Discussion Link
Description
I started a discussion about this in the FastAPI project with @Kludex, and he asked me to open this task here and link the discussion link.
However, the problem I had was after FastAPI upgraded from 0.104.1 to 0.110.0.
My application did not pass the unit tests.
I´ve tried to use version 0.104.1, and working correctly, but 0.110.0 is not working.
In this example, using FastAPI, you can see that the CancelledError occurred.
Output error:
Example Code
Python, Uvicorn & OS Version