Je suis occupé à écrire un petit serveur de jeu pour essayer flask. Le jeu expose une API via REST aux utilisateurs. Il est facile pour les utilisateurs d'effectuer des actions et d'interroger les données, mais j'aimerais servir l'API de l'utilisateur. "game world" en dehors de la app.run() boucle pour mettre à jour les entités du jeu, etc. Étant donné que Flask est si proprement implémenté, j'aimerais voir s'il y a un moyen Flask de faire cela.
Réponses
Trop de publicités?
Vos threads supplémentaires doivent être lancés à partir de la même application qui est appelée par le serveur WSGI.
L'exemple ci-dessous crée un thread d'arrière-plan qui s'exécute toutes les 5 secondes et manipule des structures de données qui sont également disponibles pour les fonctions routées de Flask.
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:
pass
# 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() Appelez-le de Gunicorn avec quelque chose comme ça :
gunicorn -b 0.0.0.0:5000 --log-config log.conf --pid=app.pid myfile:app
Andreas Bergström
Points
968
En plus d'utiliser les threads purs ou la file d'attente Celery (notez que flask-celery n'est plus nécessaire), vous pouvez également jeter un coup d'œil à flask-apscheduler :
https://github.com/viniciuschiele/flask-apscheduler
Un exemple simple copié de 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()
Dmitry Ponyatov
Points
1
Tout d'abord, vous devez utiliser une WebSocket ou un mécanisme d'interrogation pour notifier à la partie frontale les changements qui se sont produits. J'utilise Flask-SocketIO et très heureux avec messagerie asynchrone pour mes petites applications.
Nest, vous pouvez effectuer toute la logique dont vous avez besoin dans un ou plusieurs threads séparés, et notifier le frontend par l'intermédiaire de SocketIO (Flask maintient une connexion ouverte en permanence avec chaque client frontal).
À titre d'exemple, je viens d'implémenter le rechargement de la page lors de modifications de fichiers backend :
<!doctype html>
<script>
sio = io()
sio.on('reload',(info)=>{
console.log(['sio','reload',info])
document.location.reload()
})
</script>
class App(Web, Module):
def __init__(self, V):
## flask module instance
self.flask = flask
## wrapped application instance
self.app = flask.Flask(self.value)
self.app.config['SECRET_KEY'] = config.SECRET_KEY
## `flask-socketio`
self.sio = SocketIO(self.app)
self.watchfiles()
## inotify reload files after change via `sio(reload)``
def watchfiles(self):
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class Handler(FileSystemEventHandler):
def __init__(self,sio):
super().__init__()
self.sio = sio
def on_modified(self, event):
print([self.on_modified,self,event])
self.sio.emit('reload',[event.src_path,event.event_type,event.is_directory])
self.observer = Observer()
self.observer.schedule(Handler(self.sio),path='static',recursive=True)
self.observer.schedule(Handler(self.sio),path='templates',recursive=True)
self.observer.start()
