Запустите код вместе с приложением Flask

Question

Я написал веб-интерфейс для моего приложения на Python.Это прекрасно работает при запуске export FLASK_APP=main.py с последующим flask run.Теперь я хочу, чтобы само приложение также запускалось, чтобы интерфейс мог быть полезен.

Код ниже - это мой main.py, где я вызываю фабричную функцию приложений колб.

from webinterface import create_app

if __name__ == '__main__':
    create_app()
    while(True):
        # Some code logging different things

Я хочу делать что-то внутри бесконечного цикла, однако, когда я пытаюсь запустить приложение, оно либо запускает только веб-интерфейс, либо цикл infinte, в зависимости от того, запускаю ли я его с помощью flask run или python main.py.

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

Answer 1

Возможно и иногда удобно запускать Flask в потоке вне приложения переднего плана.Есть хитрость, большая ловушка и ограничение.

Ограничение заключается в том, что это то, что вы хотите делать в «безопасной» среде (например, на своем ноутбуке для сервера локального браузера илив вашей домашней интрасети), так как он включает в себя запуск сервера разработки, что вы не хотите делать в агрессивной среде.Вы также не можете использовать автоматическую перезагрузку страницы (но вы можете включить отладку).

Подводный камень в том, что, если пользовательский интерфейс разделяет любое нетривиальное состояние (включая dicts) с приложением переднего плана, вам понадобитсяиспользовать общий threading.Lock() для защиты доступа таким образом, чтобы только один поток одновременно считывал или записывал данные.

Хитрость заключается в том, чтобы вставить ссылку на общее состояние в конфигурацию приложения после его создания, но довы запускаете его, делая что-то вроде:

def webserver(state):
    app.config['STATE'] = state
    # If running on, say, a Raspberry Pi, use 0.0.0.0 so that
    # you can connect to the web server from your intranet.
    app.run(host='0.0.0.0', use_reloader=False, debug=True)

def main():
    state = SharedState()
    web_thread = threading.Thread(target=webserver, args=(state,))
    web_thread.start()

    state.set('counter' 0)
    while True:
        # Do whatever you want in the foreground thread
        state.set('counter', state.get('counter') + 1)

class SharedState():
    def __init__(self):
        self.lock = threading.Lock()
        self.state = dict()

    def get(self, key):
        with self.lock:
            return self.state.get(key)

    def set(self, key, value):
        with self.lock:
            self.state[key] = value

Затем из функций представления Flask выполните что-то вроде

@app.route('/')
def home():
    state = app.config['STATE']
    counter = state.get(counter)
    return render_template("index.html", counter=counter)