PythonでThreadを使うflaskサンプルを作ってみた - Qiita
の練習とブラウザで操作できるようにstart,stopボタンをつけてみた。
uwsgiなどで並列プロセス化するとjobs変数共有できないので、別途ジョブキューのceleryなどを使ったプロセス作る必要がある。
app.py
from datetime import datetime
from flask import Flask, make_response
from time import sleep
import threading
app = Flask(__name__)
class MyThread(threading.Thread):
def __init__(self):
super(MyThread, self).__init__()
self.stop_event = threading.Event()
def stop(self):
self.stop_event.set()
def run(self):
try:
for _ in range(1000):
print(f'{datetime.now():%H:%M:%S}')
sleep(1)
# 定期的にフラグを確認して停止させる
if self.stop_event.is_set():
break
finally:
print('時間のかかる処理が終わりました\n')
@app.route('/')
def index():
html = """
<!DOCTYPE html>
<html>
<button type="button" id="start">start</button>
<button type="button" id="stop">stop</button>
<script type="text/javascript">
document.getElementById('start').addEventListener('click',function(){
fetch("/start/1");
},false);
document.getElementById('stop').addEventListener('click',function(){
fetch("/stop/1");
},false);
</script>
</html>
"""
return html
jobs = {}
@app.route('/start/<id>/')
def start(id):
if id in jobs:
return make_response(f'{id}は実行中です\n'), 200
t = MyThread()
t.start()
jobs[id] = t
return make_response(f'{id}の処理を受け付けました\n'), 202
@app.route('/stop/<id>/')
def stop(id):
jobs[id].stop()
del jobs[id]
return make_response(f'{id}の中止処理を受け付けました\n'), 202
@app.route('/status/<id>/')
def status(id):
if id in jobs:
return make_response(f'{id}は実行中です\n'), 200
else:
return make_response(f'{id}は実行していません\n'), 200
【脱jQuery】JavaScriptをネイティブで書くときのあれこれTips | Will Style Inc.｜神戸にあるウェブ制作会社
jQueryを使わない書き方 ajax, each, trigger, on/off, extend, deferred, animate, css編 | maesblog
flask run と uWSGI prefork
python - Flaskが応答を返した後に関数を実行する
Celery Background Tasks — Flask Documentation (2.0.x)
miguelgrinberg/flask-celery-example: This repository contains the example code for my blog article Using Celery with Flask.
celery/examples at master · celery/celery