У меня есть веб-хостинг Flask без доступа к команде cron.
Как я могу выполнять некоторые функции Python каждый час?
Планирование функции для запуска каждый час на колбе
Ответ 1
Вы можете использовать BackgroundScheduler() из APScheduler пакета (v3.5.3):
import time
import atexit
from apscheduler.schedulers.background import BackgroundScheduler
def print_date_time():
print(time.strftime("%A, %d. %B %Y %I:%M:%S %p"))
scheduler = BackgroundScheduler()
scheduler.add_job(func=print_date_time, trigger="interval", seconds=3)
scheduler.start()
# Shut down the scheduler when exiting the app
atexit.register(lambda: scheduler.shutdown())
Обратите внимание, что два из этих планировщиков будут запущены, когда Flask находится в режиме отладки. Для получения дополнительной информации, проверьте этот вопрос.
Ответ 2
Вы можете использовать APScheduler в своем приложении Flask и запускать задания через его интерфейс:
import atexit
# v2.x version - see https://stackoverflow.com/a/38501429/135978
# for the 3.x version
from apscheduler.scheduler import Scheduler
from flask import Flask
app = Flask(__name__)
cron = Scheduler(daemon=True)
# Explicitly kick off the background thread
cron.start()
@cron.interval_schedule(hours=1)
def job_function():
# Do your work here
# Shutdown your cron thread if the web process is stopped
atexit.register(lambda: cron.shutdown(wait=False))
if __name__ == '__main__':
app.run()
Ответ 3
Я немного новичок в концепции планировщиков приложений, но то, что я нашел здесь для APScheduler v3.3.1, это немного другое. Я считаю, что для новейших версий структура пакета, имена классов и т.д. Изменились, поэтому я помещаю здесь свежее решение, которое я недавно сделал, интегрированное с базовым приложением Flask:
#!/usr/bin/python3
""" Demonstrating Flask, using APScheduler. """
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask
def sensor():
""" Function for test purposes. """
print("Scheduler is alive!")
sched = BackgroundScheduler(daemon=True)
sched.add_job(sensor,'interval',minutes=60)
sched.start()
app = Flask(__name__)
@app.route("/home")
def home():
""" Function for test purposes. """
return "Welcome Home :) !"
if __name__ == "__main__":
app.run()
Я также оставляю этот Gist здесь, если у кого-то есть интерес к обновлениям для этого примера.
Вот некоторые ссылки для будущих чтений:
- Документ APScheduler: https://apscheduler.readthedocs.io/en/latest/
- daemon = True: https://docs.python.org/3.4/library/threading.html#thread-objects
Ответ 4
Вы можете попробовать использовать APScheduler BackgroundScheduler для интеграции интервального задания в ваше приложение Flask. Ниже приведен пример, использующий план и фабрику приложений (init.py):
from datetime import datetime
# import BackgroundScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask
from webapp.models.main import db
from webapp.controllers.main import main_blueprint
# define the job
def hello_job():
print('Hello Job! The time is: %s' % datetime.now())
def create_app(object_name):
app = Flask(__name__)
app.config.from_object(object_name)
db.init_app(app)
app.register_blueprint(main_blueprint)
# init BackgroundScheduler job
scheduler = BackgroundScheduler()
# in your case you could change seconds to hours
scheduler.add_job(hello_job, trigger='interval', seconds=3)
scheduler.start()
try:
# To keep the main thread alive
return app
except:
# shutdown if app occurs except
scheduler.shutdown()
Надеюсь, поможет :)
Ref:
Ответ 5
Другой альтернативой может быть использование Flask-APScheduler, который прекрасно работает с Flask, например:
- Загружает конфигурацию планировщика из конфигурации Flask,
- Загружает определения заданий из конфигурации Flask
Больше информации здесь:
Ответ 6
Полный пример с использованием расписания и многопроцессорной обработки, с включенным и выключенным управлением и параметром для run_job(), коды возврата упрощаются, а интервал устанавливается равным 10 сек., Меняется на every(2).hour.do() течение every(2).hour.do(). Расписание довольно впечатляющее, оно не дрейфует, и я никогда не видел его более 100 мс при планировании. Использование многопроцессорности вместо многопоточности, потому что у него есть метод завершения.
#!/usr/bin/env python3
import schedule
import time
import datetime
import uuid
from flask import Flask, request
from multiprocessing import Process
app = Flask(__name__)
t = None
job_timer = None
def run_job(id):
""" sample job with parameter """
global job_timer
print("timer job id={}".format(id))
print("timer: {:.4f}sec".format(time.time() - job_timer))
job_timer = time.time()
def run_schedule():
""" infinite loop for schedule """
global job_timer
job_timer = time.time()
while 1:
schedule.run_pending()
time.sleep(1)
@app.route('/timer/<string:status>')
def mytimer(status, nsec=10):
global t, job_timer
if status=='on' and not t:
schedule.every(nsec).seconds.do(run_job, str(uuid.uuid4()))
t = Process(target=run_schedule)
t.start()
return "timer on with interval:{}sec\n".format(nsec)
elif status=='off' and t:
if t:
t.terminate()
t = None
schedule.clear()
return "timer off\n"
return "timer status not changed\n"
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
Вы проверяете это, просто выпуская:
$ curl http://127.0.0.1:5000/timer/on
timer on with interval:10sec
$ curl http://127.0.0.1:5000/timer/on
timer status not changed
$ curl http://127.0.0.1:5000/timer/off
timer off
$ curl http://127.0.0.1:5000/timer/off
timer status not changed
Каждые 10 секунд таймер включен, он будет выдавать сообщение таймера на консоль:
127.0.0.1 - - [18/Sep/2018 21:20:14] "GET /timer/on HTTP/1.1" 200 -
timer job id=b64ed165-911f-4b47-beed-0d023ead0a33
timer: 10.0117sec
timer job id=b64ed165-911f-4b47-beed-0d023ead0a33
timer: 10.0102sec
Ответ 7
Возможно, вы захотите использовать механизм очереди с планировщиком, например RQ scheduler или что-то более тяжелое, как сельдерей (скорее всего, перебор).
Ответ 8
Для простого решения вы можете добавить маршрут, такой как
@app.route("/cron/do_the_thing", methods=['POST'])
def do_the_thing():
logging.info("Did the thing")
return "OK", 200
Затем добавьте задание Unix cron, которое периодически отправляет сообщения в эту конечную точку. Например, чтобы запустить его раз в минуту, в терминале введите crontab -e и добавьте crontab -e строку:
* * * * * /opt/local/bin/curl -X POST https://YOUR_APP/cron/do_the_thing
(Обратите внимание, что путь к curl должен быть завершен, так как при выполнении задания у него не будет вашего PATH. Вы можете узнать полный путь к curl в вашей системе, по which curl)
Мне нравится то, что это легко проверить работу вручную, у нее нет никаких дополнительных зависимостей, и, поскольку в ней нет ничего особенного, ее легко понять.
Безопасность
Если вы хотите защитить паролем свою работу cron, вы можете pip install Flask-BasicAuth, а затем добавить учетные данные в конфигурацию вашего приложения:
app = Flask(__name__)
app.config['BASIC_AUTH_REALM'] = 'realm'
app.config['BASIC_AUTH_USERNAME'] = 'falken'
app.config['BASIC_AUTH_PASSWORD'] = 'joshua'
Чтобы защитить паролем конечную точку задания:
from flask_basicauth import BasicAuth
basic_auth = BasicAuth(app)
@app.route("/cron/do_the_thing", methods=['POST'])
@basic_auth.required
def do_the_thing():
logging.info("Did the thing a bit more securely")
return "OK", 200
Затем, чтобы позвонить с вашей работы cron:
* * * * * /opt/local/bin/curl -X POST https://falken:[email protected]_APP/cron/do_the_thing