Monkeypatch 'print' to 'yield' for streaming output

Question

I am using a long running function from a library that's primarily called from a CLI, and the status is shown by print statements. Something like this:

def long_running_function():
    ...
    LOGGER.info("first thing done")
    ...
    LOGGER.info("second thing done")
    ...

But now I want to trigger that same function from a web page and stream those print statements to the browser.

In FastAPI it looks like I should use Server-Sent Events, but that requires a function that yields messages. Like this:

async def long_running_stream():
    while True:
        yield "data: message\n\n"

Is there a way to monkeypatch those LOGGER statements into 'yield' statements? Will that even work if I am calling long_running_function() from inside long_running_stream() ?

Or is there a better way to do this?

Answer 1

Is there a way to monkeypatch those LOGGER statements into 'yield' statements?

Basically: no. I assume you do not have control over the long_running_stream implementation. (If you do have full control, then you could change it however you want.) If you then monkey patch the LOGGER.info function and make it asynchronous, this would immediately lead to an error, since long_running_stream is not async.

SSE by themselves do no require asynchronous handling. FastAI or other libraries may generate them using async functions, but this is not required. Synchronous APIs (as used in the requests lib for instance) can also handle streams (or generate them). Async APIs are generally more suitable, of course.

I want to trigger that same function from a web page and stream those print statements to the browser.

I assume you mean, you want to trigger that function on the server-side and stream the SSE from the server to the web client/browser. Sample code for this (synchronous and asynchronous, using FastAI) can be found at for instance: https://www.workfall.com/learning/blog/how-to-stream-json-data-using-server-sent-events-and-fastapi-in-python-over-http/

That particular long_running_function can probably not be used as such. So, you may have to write your own event generator.