Proper way to return errors from MJPEG stream?

[mariannasb]

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

async def async_gen(camera: BaseCamera) -> Generator[bytes, None, None]:
    """Video streaming async generator function."""
    yield b"--frame\r\n"
    try:
        while True:
            jpeg = camera.get_frame_jpeg()
            if jpeg is None:
                app_logger.debug("No jpeg frame available")
                raise V4L2Exception("No jpeg frame available")
            else:
                yield TCPSocket.package_mjpeg(jpeg.data, jpeg.metadata)
            # need this sleep in order to be able to catch the disconnect
            await asyncio.sleep(0)
    except asyncio.CancelledError:
        app_logger.debug("Client disconnected")
    except Exception as e:
        app_logger.debug(f"Unhandled exception in async_gen: {e}")
        # how to return from here something that I can detect on `async_video_feed()`?


class MJPEGResponse(StreamingResponse):
    """Custom response for MJPEG stream"""

    media_type = "multipart/x-mixed-replace; boundary=frame"
    """: The returned media type"""


# http://10.10.150.225:5000/mjpeg?dev=/dev/qtec/video0

@router.get("/mjpeg", response_class=MJPEGResponse)
async def async_video_feed(
    params: CommonParams = Depends(),  # noqa: B008
) -> MJPEGResponse:
    """Async video streaming route.

    Put this in the src attribute of an img tag.
    """
    try:
        cam = Camera(device=params.dev, stream=True)
        res = async_gen(cam)
        # how can I detect errors in `async_gen()` from here?
        return MJPEGResponse(res)
    except V4L2Exception as e:
        raise APIV4L2Error(str(e)) from e
    except RuntimeError as e:
        raise APIInputError(str(e)) from e
    except Exception as e:
        raise APIGenError(
            f"{type(e).__name__} from {sys._getframe().f_code.co_name}(): {e}"
        ) from e


@app.exception_handler(APIError)
async def unicorn_exception_handler(request: Request, exc: APIError):
    """Return custom JSON when APIError or its children are raised"""
    response = {"error": exc.description, "message": ""}
    if len(exc.args) > 0:
        response["message"] = exc.args[0]
    app_logger.error(f"{exc.description}: {response['message']}")
    return JSONResponse(
        status_code=exc.code,
        content={"message": response},
    )

Description

I have the following route and async generator for producing a MJPEG stream and it works.

My question is how can I propagate/display errors?

Fx. if I pass an invalid video device as the parameter here: http://10.10.150.225:5000/mjpeg?dev=/dev/qtec/video0, an Exception is generated by cam = Camera(device=params.dev, stream=True) and caught in the except V4L2Exception as e: and it gets handled by my exception_handler so that I get a JSON with the error message displayed on the browser.
So far so good.

My next problem is that if the video device is busy then the generator cannot generate the stream: jpeg = camera.get_frame_jpeg() returns None and in that case I would also like to be able to show a JSONResponse error in the browser.
I have tried many different iterations and can't seem to make it work.

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.85.2

Pydantic Version

1.10.12

Python Version

Python 3.11.2

Additional Context

No response