StreamingResponse generater throws exception, how to catch it and return 400?

[Cameronsplaze]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from 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)

Description

Our 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 HTTPException(detail="This is expected, here's how you fix it", status_code=400) from the other API. On successful queries, there's too much data to store all of it in ram before returning a "normal" Response too.

The above code example as is will throw a 500 ValueError: This is an error.

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 RuntimeError: Caught handled exception, but response already started., but that's exactly what I want to do. I thought about maybe somehow getting the first page manually, and then adding it back into the streaming response if it succeeded, but it returns the object out of the get_root() method long before fetching the first page. I'm not sure that's possible.

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 ValueError: This is an error. Is there a recommended way to deal with StreamingResponses that might throw exceptions? Maybe I can extend the class somehow and add custom error handling for specific exceptions that way?

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.101.1

Pydantic Version

2.3.0

Python Version

Python 3.10.12

Additional Context

No response

[Smacl3r]

Hi, we have also faced similar issue and we decided to add some additional logic into FastAPI's StreamingResponse class stream response.
Instead of raising exception we are yielding error status code together with error message, these are then correctly returned to calling client.

class StreamingResponseWithStatusCode(StreamingResponse):
    '''
    Variation of StreamingResponse that can dynamically decide the HTTP status code, based on the returns from the content iterator (parameter 'content').
    Expects the content to yield tuples of (content: str, status_code: int), instead of just content as it was in the original StreamingResponse.
    The parameter status_code in the constructor is ignored, but kept for compatibility with StreamingResponse.
    '''
    async def stream_response(self, send: Send) -> None:
        first_chunk_content, self.status_code = await self.body_iterator.__anext__() 

        if not isinstance(first_chunk_content, bytes):
            first_chunk_content = first_chunk_content.encode(self.charset)
            
        await send(
            {
                "type": "http.response.start",
                "status": self.status_code,
                "headers": self.raw_headers,
            }
        )
        await send({"type": "http.response.body", "body": first_chunk_content, "more_body": True})

        async for chunk_content, chunk_status in self.body_iterator: 
            if chunk_status // 100 != 2:
                self.status_code = chunk_status
                await send({"type": "http.response.body", "body": b"", "more_body": False})
                return
            if not isinstance(chunk_content, bytes):
                chunk_content = chunk_content.encode(self.charset)
            await send({"type": "http.response.body", "body": chunk_content, "more_body": True})

        await send({"type": "http.response.body", "body": b"", "more_body": False})

Then it your endpoint return StreamingResponseWithStatusCode instead of StreamingResponse.
When yielding response simply add status code as additional field:
yield chunk, status_code

Needed this to create a streaming proxy and it was useful to return original error code + chunk with error information.
Let me know if this helped or that it might have some drawbacks.

[rossgray]

Thanks, I found this code helpful in a scenario I ran into too

[sandangel]

Hi, I also run into this. Can you also share how you send the error message to user?

[Smacl3r]

Since I am using StreamingResponse as a proxy for another request it depends how that error is returned by the service I am calling.
In my situation it is returned in the first chunk as 'error'.

Otherwise you can just send your error message in the first yield .

Code snippet from working code:

async with client.stream(
                method='POST',
                url=url,
                headers=headers,
                json=body,
                params=query_params
            ) as response:
                async for chunk in response.aiter_bytes():
                    chunk_str = chunk.decode('utf-8')
                    if not response.is_success:
                        logging.error(f'{chunk_str}')
                        yield chunk, response.status_code

[SidJain1412]

@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 :)
edit: My bad, this is more of a starlette issue than fastapi specifically

[Kludex]

The above code example as is will throw a 500 ValueError: This is an error.

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 recommended by the spec).

Tip

Also, I help here because I want, I'm not being paid for such a thing.
If you want to demand attention, the least you can do is sponsor me.

[SidJain1412]

Apologies for the tone, I meant to request your expertise not demand your help, really appreciate your time 🙏
And thanks for the helpful answer!

[RamiAwar]

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.
This way you don't need to modify the StreamingResponse class or do anything fancy.

Hope this helps someone!

[Mastermind-U]

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))

[Saran33]

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 (chunk, status_code) for successful chunks - only when raising an error response do we need to include the status_code.
The error event format in this example matches the OpenAI API format, so this will work if you're wrapping the OpenAI completions endpoint and calling it from an OpenAI client library. For example, if calling your streaming endpoint with the openai Python lib, it would raise openai.APIError with your error message.