Запустить код после запуска фляги

Моя цель - заставить произвольный код работать после запуска моего приложения 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.

Ответ 1

Если вам нужно выполнить какой-то код после запуска флеш-приложения, но строго до первого запроса, даже не запускаться при выполнении первого запроса, так как @app.before_first_request может обрабатывать, вы должны использовать Flask_Script, поскольку CESCO сказал, но вы можете подклассифицировать сервер классов и перезаписать метод __ вызова __, вместо того чтобы перезаписать команду runerver с помощью @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()

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

Ответ 2

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

# 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()

Ответ 3

Я только что сделал (в 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 делает...

Ответ 4

Мне не очень нравится ни один из методов, упомянутых выше, потому что вам не нужен 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().

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

Ответ 5

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

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

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