How to add a background stream to a flask?

Question

I am busy writing a small game server to try out a flask. The game provides an API through REST for users. It is easy for users to perform actions and request data, however, I would like to serve the “game world” outside the app.run () loop to update game entities, etc. Given that Flask is so cleanly implemented, I would like to see if there is a way for Flask to do this.

Answer 1

Your additional threads must be initiated from the same application that is called by the WSGI server.

The following example creates a background thread that runs every 5 seconds and manages the data structures that are also available for routed Flask functions.

import threading import atexit from flask import Flask POOL_TIME = 5 #Seconds # variables that are accessible from anywhere commonDataStruct = {} # lock to control access to variable dataLock = threading.Lock() # thread handler yourThread = threading.Thread() def create_app(): app = Flask(__name__) def interrupt(): global yourThread yourThread.cancel() def doStuff(): global commonDataStruct global yourThread with dataLock: # Do your stuff with commonDataStruct Here # Set the next thread to happen yourThread = threading.Timer(POOL_TIME, doStuff, ()) yourThread.start() def doStuffStart(): # Do initialisation stuff here global yourThread # Create your thread yourThread = threading.Timer(POOL_TIME, doStuff, ()) yourThread.start() # Initiate doStuffStart() # When you kill Flask (SIGTERM), clear the trigger for the next thread atexit.register(interrupt) return app app = create_app() 

Call it from Gunicorn with something like this:

 gunicorn -b 0.0.0.0:5000 --log-config log.conf --pid=app.pid myfile:app 

Answer 2

It seems like a hacker way to do this , but I don't think it is technically supported.

I also found this answer , which talks about using a celery bulb for this.

Answer 3

In addition to using a clean thread or a line of celery (note that celery cones are no longer required), you can also take a look at the approcessor flask:

https://github.com/viniciuschiele/flask-apscheduler

A simple example copied from https://github.com/viniciuschiele/flask-apscheduler/blob/master/examples/jobs.py :

 from flask import Flask from flask_apscheduler import APScheduler class Config(object): JOBS = [ { 'id': 'job1', 'func': 'jobs:job1', 'args': (1, 2), 'trigger': 'interval', 'seconds': 10 } ] SCHEDULER_API_ENABLED = True def job1(a, b): print(str(a) + ' ' + str(b)) if __name__ == '__main__': app = Flask(__name__) app.config.from_object(Config()) scheduler = APScheduler() # it is also possible to enable the API directly # scheduler.api_enabled = True scheduler.init_app(app) scheduler.start() app.run()