Запустите код после запуска колбы

Question

Моя цель - запустить произвольный код после запуска приложения Flask. Вот что у меня есть:

def run():
    from webapp import app
    app.run(debug=True, use_reloader=False)

В идеале я мог бы просто сделать это:

def run():
    from webapp import app
    app.run(debug=True, use_reloader=False)
    some_code()

Но код не продолжается мимо app.run(), так что some_code() никогда не запускается.

Решение, над которым я сейчас работаю, состоит в том, чтобы запустить some_code() в отдельном потоке от app.run(), создать функцию запроса перед первым запросом, которая устанавливает это:

app.is_running = True

Затем заставьте some_code() выполнить базовый запрос к приложению, чтобы выполнялся код "перед первым запросом". Это довольно запутанно и будет трудно документировать. Я бы предпочел использовать параметр app.is_running, который уже представлен в Flask, или использовать @app.after_server_start декоратор, но, насколько мне известно, ни один из них не существует.

Помогите мне сделать этот код лучше?

---

Посмертно: каждый раз, когда я думаю об этой проблеме, мне хочется, чтобы @app.after_server_start декоратор существовал.

Answer 1

Если вам нужно выполнить некоторый код после запуска вашего приложения-колбы, но строго перед первым запросом, даже не запускаться при выполнении первого запроса, как может обрабатывать @app.before_first_request, вы должны использовать Flask_Script, как сказал CESCO, но вы мог бы создать подкласс класса Server и переписать метод __call__, вместо того, чтобы перезаписать команду runserver с @manager.command:

from flask import Flask
from flask_script import Manager, Server

def custom_call():
    #Your code
    pass

class CustomServer(Server):
    def __call__(self, app, *args, **kwargs):
        custom_call()
        #Hint: Here you could manipulate app
        return Server.__call__(self, app, *args, **kwargs)

app = Flask(__name__)
manager = Manager(app)

# Remeber to add the command to your Manager instance
manager.add_command('runserver', CustomServer())

if __name__ == "__main__":
    manager.run()

Таким образом, вы не переопределяете параметры по умолчанию команды runserver.

Answer 2

Я только что сделал (в main.py выполнено с python main.py):

with app.app_context():
    from module import some_code()
    some_code()

def run():
    from webapp import app
    app.run(debug=True, use_reloader=False)

Это работало для меня без более полных ответов, предложенных выше. В моем случае some_code() инициализирует кеш через flask_caching.Cache,

Но это, вероятно, зависит от того, что именно some_code делается...

Answer 3

Мне не очень нравится ни один из методов, упомянутых выше, потому что вам не нужен Flask-Script, и не все проекты уже будут использовать Flask-Script.

Самый простой способ - создать собственный подкласс Flask. Где вы строите свое приложение с Flask(__name__)Вы просто добавили бы свой собственный класс и использовали бы его вместо этого.

def do_something():
  print('MyFlaskApp is starting up!')


class MyFlaskApp(Flask):
  def run(self, host=None, port=None, debug=None, load_dotenv=True, **options):
    if not self.debug or os.getenv('WERKZEUG_RUN_MAIN') == 'true':
      with self.app_context():
        do_something()
    super(MyFlaskApp, self).run(host=host, port=port, debug=debug, load_dotenv=load_dotenv, **options)


app = MyFlaskApp(__name__)
app.run()

Конечно, это не запускается после запуска, но прямо перед run() наконец-то называется. С контекстом приложения вы должны иметь возможность делать все, что вам может понадобиться с базой данных, или все, что требует контекста приложения. Это должно работать с любым сервером (uwsgi, gunicorn и т. Д.).

Если вам нужно do_something() чтобы быть неблокирующим, вы можете просто threading.Thread(target=do_something).start() вместо.

Условный оператор должен предотвратить двойной вызов при использовании режима отладки / перегрузчика.

Answer 4

Используйте Flask-Script для запуска вашего приложения, затем перезапишите класс / метод runserver следующим образом

# manage.py

from flask.ext.script import Manager

from myapp import app

manager = Manager(app)

def crazy_call():
    print("crazy_call")

@manager.command
def runserver():
    app.run()
    crazy_call()

if __name__ == "__main__":
    manager.run()

Answer 5

Если вы хотите запустить набор команд (как обычное приложение Py) после запуска фляги, используйте библиотеку с несколькими процессорами (например, многопроцессорность). В этом решении вы можете иметь приложение API/ веб-приложение рядом с системной программой.

import flask
from flask import request, jsonify,make_response
import time
import multiprocessing

app = flask.Flask('__name__')

def API(Conf):
   print('In API selction')
   app.run(host='0.0.0.0', port=1337,)
if __name__ == "__main__":
   config = {"Something":"SomethingElese"}
   p = multiprocessing.Process(target=API, args=(Conf,))
   p.start()
   #time.sleep(3)
   print('After Flask run')

Примечание: приведенный выше код - это всего лишь образец / идея. Может быть какая-то ошибка. (Но я использовал это решение на производстве, и все в порядке.)

PS: (с моей точки зрения) проблема этого решения: больше сложностей в разделе отладки.

Answer 6

Я использовал Flask, чтобы проверить, правильно ли взаимодействуют веб-сервис и клиентское приложение. (То есть в рамках автоматического регрессионного теста, чтобы убедиться, что подключение веб-интерфейса к внутренней утилите было протоколом.)

Я запустил flask в параллельном потоке, например:

      import flask, requests, multiprocessing

app = flask.Flask(__name__)

@app.route("/"):
def frontpage():
    return "Hello World"

server = multiprocessing.Process(target=app.run)
try:
    server.start()

    # Is time.sleep(n) needed here to avoid a potential race condition?

    assert requests.get("http://127.0.0.1:5000/").text == "Hello World"

finally:
    server.terminate()
    server.join()

Answer 7

Я столкнулся с той же самой проблемой в моем приложении фляги. Я хотел запустить планировщик при запуске приложения, который будет запускать некоторые задания через регулярные промежутки времени. Поскольку я развертываю свои приложения в докер-контейнерах, в итоге я добавил конечную точку "проверки работоспособности", которая просто возвращает 200, и в моем dockerfile настроил эту конечную точку:

HEALTHCHECK CMD curl --fail http://localhost:8080/alive || exit 1

Поведение по умолчанию - выполнять эту команду каждые 30 секунд, и первый запуск удобно запускает мой метод init(). https://docs.docker.com/engine/reference/builder/