I'd like to implement a flask based web server which continuously reads data (e.g. from a keyboard, but could be any "outside world" device) and replies to GET requests based on that data.
An easy to imagine use case would be a web service telling whether or not a certain key on the keyboard is currently pressed.
A naive approach would be to just start a thread, which continuously reads keyboard events and modifies a data structure based on those events and in parallel run a flask app using the data structure.
I'm now wondering if it's possible to run those tasks in parallel using asyncio instead of threads (for safety reasons and out of curiosity)
I found lots of questions and answers regarding flask and async requests, but those (I think) handle the case when you have multiple requests in parallel you want to reply to asynchronously. (e.g. this one: Making an asynchronous task in Flask). In contrast I don't need asynchronously handled requests but a continuously running task in parallel to the web server.
An approach I would think of could look like this:
import asyncio
from flask import Flask, Response
app = Flask(__name__, static_url_path="/")
class Keyboard:
def was_pressed(self, key):
if key in self.pressed:
self.pressed.remove(a)
return True
return False
async def handle_event(self, key):
self.pressed.add(key)
keyboard = Keyboard()
@app.route("/waspressed/<path:key>")
def waspressed(key):
return Response(str(keyboard.was_pressed(key)))
async def continuously_update_keyboard_state():
while True:
keyboard.handle_event(await keyboard_event())
async def main():
await asyncio.gather(
continuously_update_keyboard_state(),
app.run(host="0.0.0.0", port="9000"))
if __name__ == "__main__":
asyncio.run(main())
but flask.Flask.run() is not implemented using async. So that approach is not possible, correct? Can I somehow wrap it? Am I on a completely wrong track here?