make_test_body_with_headers(I get a non-empty body),
await request.filesreturns an empty dict inside my webapp code. I think this is related with the multipart header value not containing the boundary?
I've written about how to use postgres with Quart, specifying my preferred libraries (Databases & asyncpg) and how to convert types automatically,
Would be good to know if you think this is better as an extension
Hi @pgjones I am stumped with the following. I'm trying to pass a
cursor_id to a websocket view, like so:
@home_app.websocket("/ws") @login_required async def ws(): dbc = current_app.dbc cursor_id = int(request.args.get("cursor_id"))
But when I try to read the argument I get the error: "Attempt to access request outside of a relevant context". How can I pass that value from the template to the websocket endpoint?
FileStorage.streamnot being an asyncio object, or would we be locked waiting for the (currently) 16KB chunks of data on every iteration? I think those would probably be fine considering the kernel can cache possible future file reads, but I'm not sure.
queue.get()also never worked
This is the code on the docs
connected_websockets = set() def collect_websocket(func): @wraps(func) async def wrapper(*args, **kwargs): global connected_websockets queue = asyncio.Queue() connected_websockets.add(queue) try: return await func(queue, *args, **kwargs) finally: connected_websockets.remove(queue) return wrapper
This creates an empty queue, and then adds it to the
set, but the queue is completely empty.
len(connected_websockets)would indicate how many connected websockets you have
@pgjones sorry for the late reply, but here's my code:
from quart import Quart, websocket from functools import wraps import asyncio app = Quart(__name__) connected_websockets = set() def collect_websocket(func): @wraps(func) async def wrapper(*args, **kwargs): global connected_websockets queue = asyncio.Queue() connected_websockets.add(queue) print(connected_websockets) try: return await func(queue, *args, **kwargs) finally: connected_websockets.remove(queue) return wrapper async def broadcast(message): for queue in connected_websockets: queue.put_nowait("New connection") @app.websocket('/') @collect_websocket async def ws(queue): while True: data = await websocket.receive() print(data) await broadcast(data) if __name__ == "__main__": app.run(host="0.0.0.0", port=8080)
You can ignore the
host="0.0.0.0", that's just there because it's on replit