Как вызвать функцию после оператора возврата в Flask

Question

У меня есть 2 функции.

1-я функция сохраняет данные, полученные в списке, а 2-я функция записывает данные в файл CSV.

Я использую Flask.Когда бы ни был вызван веб-сервис, он будет хранить данные и отправлять на них ответ, как только он отправляет ответ, он запускает вторую функцию.

Мой код :

from flask import Flask, flash, request, redirect, url_for, session
import json

app = Flask(__name__)

arr = []

@app.route("/test", methods=['GET','POST'])
def check():
    arr.append(request.form['a'])
    arr.append(request.form['b'])
    res = {'Status': True}
    return json.dumps(res)

def trigger():
    df = pd.DataFrame({'x': arr})
    df.to_csv("docs/xyz.csv", index=False)
    return 

Очевидно, что 2-я функция не вызывается.

Есть ли способ достичь этого?

PS: Моя проблема в реальной жизни отличается, когда функция trigger отнимает много времени, и я не хочу, чтобы пользователь ждал, пока она завершит выполнение.

Answer 1

Вы можете попробовать использовать потоковую передачу.См. Следующий пример:

import time
from flask import Flask, Response

app = Flask(__name__)

@app.route('/')
def main():
    return '''<div>start</div>
    <script>
        var xhr = new XMLHttpRequest();
        xhr.open('GET', '/test', true);
        xhr.onreadystatechange = function(e) {
            var div = document.createElement('div');
            div.innerHTML = '' + this.readyState + ':' + this.responseText;
            document.body.appendChild(div);
        };
        xhr.send();
    </script>
    '''

@app.route('/test')
def test():
    def generate():
        app.logger.info('request started')
        for i in range(5):
            time.sleep(1)
            yield str(i)
        app.logger.info('request finished')
        yield ''
    return Response(generate(), mimetype='text/plain')

if __name__ == '__main__':
    app.run('0.0.0.0', 8080, True)

Вся магия в этом примере в genarator, где вы можете запустить ответные данные, после того, как сделаете некоторое количество персонала и вывести пустые данные для завершения вашего потока.

Подробнее см.http://flask.pocoo.org/docs/patterns/streaming/.

Answer 2

Одним из решений было бы иметь фоновый поток, который будет наблюдать за очередью.Вы помещаете свои CSV-данные в очередь, и фоновый поток будет их использовать.Вы можете запустить такой поток перед первым запросом:

import threading
from multiprocessing import Queue

class CSVWriterThread(threading.Thread):
    def __init__(self, *args, **kwargs):
        threading.Thread.__init__(self, *args, **kwargs)
        self.input_queue = Queue()

    def send(self, item):
        self.input_queue.put(item)

    def close(self):
        self.input_queue.put(None)
        self.input_queue.join()

    def run(self):
        while True:
            csv_array = self.input_queue.get()
            if csv_array is None:
                break

            # Do something here ...
            df = pd.DataFrame({'x': csv_array})
            df.to_csv("docs/xyz.csv", index=False)


            self.input_queue.task_done()
            time.sleep(1)
        # Done
        self.input_queue.task_done()
        return

@app.before_first_request
def activate_job_monitor():
    thread = CSVWriterThread()
    app.csvwriter = thread
    thread.start()

И в своем коде поместить сообщение в очередь перед возвратом:

@app.route("/test", methods=['GET','POST'])
def check():
    arr.append(request.form['a'])
    arr.append(request.form['b'])
    res = {'Status': True}
    app.csvwriter.send(arr)
    return json.dumps(res)

Answer 3

PS: Моя проблема в реальной жизни отличается, когда функция триггера отнимает много времени, и я не хочу, чтобы пользователь ждал, пока она завершит выполнение.

Рассмотрите возможность использования сельдерея, которая создана для той самой проблемы, которую вы пытаетесь решить.Из документов:

Celery - это простая, гибкая и надежная распределенная система для обработки огромного количества сообщений и обеспечения операций с инструментами, необходимыми для обслуживания такой системы.

Я рекомендую вам интегрировать сельдерей с вашим приложением для колб, как описано здесь .Ваш trigger метод станет простой задачей сельдерея, которую вы можете выполнить, не беспокоясь о длительном времени ответа.

Answer 4

Я на самом деле работаю над другим интересным случаем на моей стороне, где я передаю работу работнику Python, который отправляет работу в очередь redis.Есть несколько отличных блогов, использующих Redis с Flask, вам нужно убедиться, что Redis работает (может подключаться через порт 6379)

Рабочий будет выглядеть примерно так:

import os

import redis
from rq import Worker, Queue, Connection

listen = ['default']

redis_url = os.getenv('REDISTOGO_URL', 'redis://localhost:6379')

conn = redis.from_url(redis_url)

if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(list(map(Queue, listen)))
        worker.work()

ВВ моем примере у меня есть функция, которая запрашивает базу данных для использования, и, поскольку это может быть длительным процессом, я передаю ее работнику (работающему как отдельный скрипт)

def post(self):

    data = Task.parser.parse_args()

    job = q.enqueue_call(
        func=migrate_usage, args=(my_args),
        result_ttl=5000
    )
    print("Job ID is: {}".format(job.get_id()))
    job_key = job.get_id()

    print(str(Job.fetch(job_key, connection=conn).result))

    if job:
        return {"message": "Job : {} added to queue".format(job_key)}, 201

Кредит из-за следующей статьи:

https://realpython.com/flask-by-example-implementing-a-redis-task-queue/#install-requirements