StreamingResponse generater throws exception, how to catch it and return 400? #10138
-
First Check
Commit to Help
Example Codefrom fastapi import FastAPI, Request, HTTPException
from fastapi.responses import Response, JSONResponse, StreamingResponse
import uvicorn
app = FastAPI()
pages = [
{"page_1": 1},
{"page_2": 2},
{"page_3": 3},
{"page_4": 4},
]
def yield_geojson():
for page in pages:
raise ValueError("This is an error")
yield str(page)
@app.get("/")
async def get_root():
print("Creating response_info")
response_info = {
'media_type': 'application/geojson; charset=utf-8',
'content': yield_geojson(),
'headers': {
'Access-Control-Expose-Headers': 'Content-Disposition',
'Access-Control-Allow-Origin': '*'
}
}
try:
print("Starting StreamingResponse")
return StreamingResponse(**response_info)
except ValueError as exc:
print("Caught error") # This doesn't catch, the method returns before fetching the first page.
return JSONResponse(status_code=400, content={"message": str(exc)})
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=3000) DescriptionOur API sort-of acts like a proxy. We take search params, and use them to page through another API's results. We have a generator we create that converts the params from what we expect, to what the external API expects, and also sets appropriate defaults for our use case. It then uses those to page through results and return them to us, which we return to the client. The problem is, if our query contains a bad param, the generator that normally returns pages will instead throw a 400 bad request. (Represented by the ValueError above). And it's not necessarily the first page that'll throw in some circumstances too. I can't figure out how to forward this error message The above code example as is will throw a 500 I've also tried adding an exception handler: @app.exception_handler(ValueError)
async def handle_error(request: Request, error: HTTPException):
response = {
"error": {
"type": "ERROR",
"report": error.detail,
}
}
return JSONResponse(
content=response,
status_code=error.status_code,
) This will throw a From this comment, I also tried adding a custom route to give us some error handling: from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
from fastapi.responses import JSONResponse, StreamingResponse
import uvicorn
class ExceptionRoute(APIRoute):
def get_route_handler(self):
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
try:
response: Response = await original_route_handler(request)
print("Created a valid response") # This runs, above never throws
return response
except ValueError as e:
print("HIT the error") # This never runs....
return JSONResponse(
content={"detail": "this is how you deal with the error"}, status_code=400
)
return custom_route_handler
app = FastAPI()
router = APIRouter(route_class=ExceptionRoute)
pages = [
{"page_1": 1},
{"page_2": 2},
{"page_3": 3},
{"page_4": 4},
]
def yield_geojson():
for page in pages:
raise ValueError("This is an error")
yield str(page)
@router.get("/")
async def root():
print("Creating response_info")
response_info = {
'media_type': 'application/geojson; charset=utf-8',
'content': yield_geojson(),
'headers': {
'Access-Control-Expose-Headers': 'Content-Disposition',
'Access-Control-Allow-Origin': '*'
}
}
return StreamingResponse(**response_info)
app.include_router(router)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=3000) And this is back to the Operating SystemLinux Operating System DetailsNo response FastAPI Version0.101.1 Pydantic Version2.3.0 Python VersionPython 3.10.12 Additional ContextNo response |
Beta Was this translation helpful? Give feedback.
Replies: 6 comments 4 replies
-
Hi, we have also faced similar issue and we decided to add some additional logic into FastAPI's StreamingResponse class stream response.
Then it your endpoint return Needed this to create a streaming proxy and it was useful to return original error code + chunk with error information. |
Beta Was this translation helpful? Give feedback.
-
@Kludex this is an issue that I feel requires attention as more people build Streaming APIs using FastAPI. Is there a proper solution instead of overriding the StreamingResponse class? Appreciate your help :) |
Beta Was this translation helpful? Give feedback.
-
No, the code above doesn't throw a 500. It throws a 200. This may be a bit confusing, but response headers, and status code are sent in the beginning (before processing the stream). You may benefit from something called HTTP trailers, which are supported by uvicorn. You can see more background about it on: encode/starlette#1876 (tentative PR against Starlette, that I didn't go through). Well, to summarize... No way to catch errors and send a different error code using the Response classes from Starlette. You can create your custom one and only send the Tip Also, I help here because I want, I'm not being paid for such a thing. |
Beta Was this translation helpful? Give feedback.
-
One simple solution I've used in https://dataline.app is just wrapping the generator with another generator that captures these errors and packages them nicely 🙂 From https://github.com/RamiAwar/dataline: def stream_event_str(event: str, data: str) -> str:
# https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
event_str = f"event: {event}"
data_str = f"data: {data}"
return f"{event_str}\n{data_str}\n\n"
async def generate_with_errors(generator: AsyncGenerator[str, None]) -> AsyncGenerator[str, None]:
try:
async for chunk in generator:
yield chunk
except UserFacingError as e:
logger.exception("Error in conversation query generator")
yield stream_event_str(QueryStreamingEventType.ERROR.value, str(e)) Then you can handle this error on the frontend. Hope this helps someone! |
Beta Was this translation helpful? Give feedback.
-
If you are not streaming endless media, you can use this example, I found in httpx docs: import httpx
from starlette.background import BackgroundTask
from fastapi.responses import StreamingResponse
# NOTE: use dependency injection instead of this global
client = httpx.AsyncClient()
@router.get('')
async def home() -> bytes:
req = client.build_request("GET", "https://www.example.com/")
response = await client.send(req, stream=True)
if response.status_code != 200:
raise HTTPException(400)
return StreamingResponse(response.aiter_text(), background=BackgroundTask(response.aclose)) |
Beta Was this translation helpful? Give feedback.
-
Building on @Smacl3r's answer, to include error events with error messages, and handling of any unhandled exceptions: import json
from collections.abc import AsyncIterator
import structlog
from fastapi.responses import StreamingResponse
from starlette.types import Send
logger = structlog.stdlib.get_logger()
class StreamingResponseWithStatusCode(StreamingResponse):
"""
Variation of StreamingResponse that can dynamically decide the HTTP status code,
based on the return value of the content iterator (parameter `content`).
Expects the content to yield either just str content as per the original `StreamingResponse`
or else tuples of (`content`: `str`, `status_code`: `int`).
"""
body_iterator: AsyncIterator[str | bytes]
response_started: bool = False
async def stream_response(self, send: Send) -> None:
more_body = True
try:
first_chunk = await self.body_iterator.__anext__()
if isinstance(first_chunk, tuple):
first_chunk_content, self.status_code = first_chunk
else:
first_chunk_content = first_chunk
if isinstance(first_chunk_content, str):
first_chunk_content = first_chunk_content.encode(self.charset)
await send(
{
"type": "http.response.start",
"status": self.status_code,
"headers": self.raw_headers,
}
)
self.response_started = True
await send(
{
"type": "http.response.body",
"body": first_chunk_content,
"more_body": more_body,
}
)
async for chunk in self.body_iterator:
if isinstance(chunk, tuple):
content, status_code = chunk
if status_code // 100 != 2:
# An error occurred mid-stream
if not isinstance(content, bytes):
content = content.encode(self.charset)
more_body = False
await send(
{
"type": "http.response.body",
"body": content,
"more_body": more_body,
}
)
return
else:
content = chunk
if isinstance(content, str):
content = content.encode(self.charset)
more_body = True
await send(
{
"type": "http.response.body",
"body": content,
"more_body": more_body,
}
)
except Exception as exc:
await logger.aexception("unhandled_streaming_error", error=str(exc))
more_body = False
error_resp = {"error": {"message": "Internal Server Error"}}
error_event = f"event: error\ndata: {json.dumps(error_resp)}\n\n".encode(
self.charset
)
if not self.response_started:
await send(
{
"type": "http.response.start",
"status": 500,
"headers": self.raw_headers,
}
)
await send(
{
"type": "http.response.body",
"body": error_event,
"more_body": more_body,
}
)
if more_body:
await send({"type": "http.response.body", "body": b"", "more_body": False}) import json
from collections.abc import AsyncGenerator
from typing import Any
from openai import APIStatusError as OpenAIStatusError
from structlog.stdlib import BoundLogger
async def handle_openai_stream_exc(
exc, logger: BoundLogger
) -> AsyncGenerator[tuple[str, int], Any]:
if isinstance(exc, OpenAIStatusError):
await logger.aerror("SOME_ERROR", error=exc)
error_resp = {
"error": {
"message": exc.message,
"type": exc.type,
"param": exc.param,
"code": exc.code,
}
}
yield f"event: error\ndata: {json.dumps(error_resp)}\n\n", exc.status_code
else:
await logger.aexception("SOME_ERROR", error=str(exc))
error_resp = {"error": {"message": "Internal Server Error"}}
yield f"event: error\ndata: {json.dumps(error_resp)}\n\n", 500 In this approach we don't need to yield a tuple of |
Beta Was this translation helpful? Give feedback.
No, the code above doesn't throw a 500. It throws a 200.
This may be a bit confusing, but response headers, and status code are sent in the beginning (before processing the stream).
You may benefit from something called HTTP trailers, which are supported by uvicorn. You can see more background about it on: encode/starlette#1876 (tentative PR against Starlette, that I didn't go through).
Well, to summarize... No way to catch errors and send a different error code using the Response classes from Starlette. You can create your custom one and only send the
http.response.start
AFTER processing the body (which is also…