Ensure generators are closed

[pgjones]

import asyncio
from quart import Quart, Response

app = Quart(__name__)

@app.route('/')
async def sub():
        response = Response(
            data_stream(),
            mimetype="text/event-stream",
        )
        response.timeout = None
        return response

async def data_stream():
    try:
        while True:
            print("Sending Data")
            if False:
                yield "data: {ok: true}\n\n"
            await asyncio.sleep(1)
    finally:
        print("Exited Gen")
    
app.run()

[sungeer]

ha, please follow me :

import asyncio
from quart import Quart, make_response, stream_with_context
from datetime import datetime

app = Quart(__name__)

@app.route('/')
async def sub():
    @stream_with_context
    async def data_stream():
        try:
            while True:
                print("Sending Data")
                yield "data: {ok: true}\n\n".encode()
                await asyncio.sleep(1)
        finally:
            print("Exited Gen")

    response = await make_response(data_stream())
    response.timeout = None
    response.headers['Content-Type'] = 'text/event-stream'
    return response

if __name__ == "__main__":
    app.run()