Frasco por exemplo - Implementando uma fila de tarefas do Redis

• 12/02/2020:atualizado para o Python versão 3.8.1, bem como para as versões mais recentes do Redis, Python Redis e RQ. Veja abaixo para detalhes. Mencione um bug na versão RQ mais recente e forneça uma solução. Resolvido o bug http antes de https.
• 22/03/2016:atualizado para o Python versão 3.5.1, bem como para as versões mais recentes do Redis, Python Redis e RQ. Veja abaixo os detalhes.
• 22/02/2015:Adicionado suporte para Python 3.

1. Parte um:configurar um ambiente de desenvolvimento local e, em seguida, implantar um ambiente de preparação e de produção no Heroku.
2. Parte dois:configure um banco de dados PostgreSQL junto com SQLAlchemy e Alembic para lidar com migrações.
3. Parte três:adicione a lógica de back-end para extrair e, em seguida, processar a contagem de palavras de uma página da Web usando as bibliotecas de solicitações, BeautifulSoup e Natural Language Toolkit (NLTK).
4. Parte quatro:Implemente uma fila de tarefas do Redis para lidar com o processamento de texto. (atual )
5. Parte cinco:configure o Angular no front-end para pesquisar continuamente o back-end para ver se o processamento da solicitação foi concluído.
6. Parte seis:enviar para o servidor de teste no Heroku - configurando o Redis e detalhando como executar dois processos (web e worker) em um único Dyno.
7. Parte sete:atualize o front-end para torná-lo mais fácil de usar.
8. Parte oito:crie uma diretiva angular personalizada para exibir um gráfico de distribuição de frequência usando JavaScript e D3.

Requisitos de instalação

• Redis (5.0.7)
• Python Redis (3.4.1)
• RQ (1.2.2) - uma biblioteca simples para criar uma fila de tarefas

$ redis-server

$ cd flask-by-example
$ python -m pip install redis==3.4.1 rq==1.2.2
$ python -m pip freeze > requirements.txt

Configurar o trabalhador

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

$ cd flask-by-example
$ python worker.py
17:01:29 RQ worker started, version 0.5.6
17:01:29
17:01:29 *** Listening on default...

Atualizar app.py

from rq import Queue
from rq.job import Job
from worker import conn

app = Flask(__name__)
app.config.from_object(os.environ['APP_SETTINGS'])
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)

q = Queue(connection=conn)

from models import *

def count_and_save_words(url):

    errors = []

    try:
        r = requests.get(url)
    except:
        errors.append(
            "Unable to get URL. Please make sure it's valid and try again."
        )
        return {"error": errors}

    # text processing
    raw = BeautifulSoup(r.text).get_text()
    nltk.data.path.append('./nltk_data/')  # set the path
    tokens = nltk.word_tokenize(raw)
    text = nltk.Text(tokens)

    # remove punctuation, count raw words
    nonPunct = re.compile('.*[A-Za-z].*')
    raw_words = [w for w in text if nonPunct.match(w)]
    raw_word_count = Counter(raw_words)

    # stop words
    no_stop_words = [w for w in raw_words if w.lower() not in stops]
    no_stop_words_count = Counter(no_stop_words)

    # save the results
    try:
        result = Result(
            url=url,
            result_all=raw_word_count,
            result_no_stop_words=no_stop_words_count
        )
        db.session.add(result)
        db.session.commit()
        return result.id
    except:
        errors.append("Unable to add item to database.")
        return {"error": errors}


@app.route('/', methods=['GET', 'POST'])
def index():
    results = {}
    if request.method == "POST":
        # this import solves a rq bug which currently exists
        from app import count_and_save_words

        # get url that the person has entered
        url = request.form['url']
        if not url[:8].startswith(('https://', 'http://')):
            url = 'http://' + url
        job = q.enqueue_call(
            func=count_and_save_words, args=(url,), result_ttl=5000
        )
        print(job.get_id())

    return render_template('index.html', results=results)

job = q.enqueue_call(
    func=count_and_save_words, args=(url,), result_ttl=5000
)
print(job.get_id())

Observação: Precisamos importar as count_and_save_words função em nossa função index pois o pacote RQ atualmente tem um bug, onde não encontra funções no mesmo módulo.

Obter resultados

@app.route("/results/<job_key>", methods=['GET'])
def get_results(job_key):

    job = Job.fetch(job_key, connection=conn)

    if job.is_finished:
        return str(job.result), 200
    else:
        return "Nay!", 202

# save the results
try:
    from models import Result
    result = Result(
        url=url,
        result_all=raw_word_count,
        result_no_stop_words=no_stop_words_count
    )
    db.session.add(result)
    db.session.commit()
    return result.id

@app.route("/results/<job_key>", methods=['GET'])
def get_results(job_key):

    job = Job.fetch(job_key, connection=conn)

    if job.is_finished:
        result = Result.query.filter_by(id=job.result).first()
        results = sorted(
            result.result_no_stop_words.items(),
            key=operator.itemgetter(1),
            reverse=True
        )[:10]
        return jsonify(results)
    else:
        return "Nay!", 202

from flask import jsonify

[
  [
    "Python", 
    315
  ], 
  [
    "intermediate", 
    167
  ], 
  [
    "python", 
    161
  ], 
  [
    "basics", 
    118
  ], 
  [
    "web-dev", 
    108
  ], 
  [
    "data-science", 
    51
  ], 
  [
    "best-practices", 
    49
  ], 
  [
    "advanced", 
    45
  ], 
  [
    "django", 
    43
  ], 
  [
    "flask", 
    41
  ]
]

O que vem a seguir?