Executando processos em background com Django e Celery

Se você já desenvolveu alguma aplicação em Django, provavelmente já se deparou em algum processo com tempo de execução mais longo podendo levar minutos ou horas e, consequentemente, foi visitado pelo famoso Timeout. Isso é bastante comum em aplicações web que empenham papéis importantes e possuem um fluxo de trabalho mais extenso e demorado o qual é necessário uma execução assíncrona e não blocantes de suas tarefas. Um exemplo bem comum é: as gerações de relatórios que necessitam buscar os dados em diversas APIs, consolidar esses dados e gerar uma planilha com as informações obtidas. Pensando nisso, o Celery surgiu para tirar esse Karma. Mas afinal, o que é esse tal de Celery?

Celery

Celery logo

Segundo os criadores, Celery é um sistema distribuído sólido, simples e flexível que lhe permite executar grande números de mensagens de forma assíncrona. É um task queue com foco no processamento em tempo real, além de suportar o agendamento de tarefas.

Celery é Open Source e está sob a licença BSD.

Dessa forma, irei abordar alguns tópicos importantes para o entendimento de como tudo funciona.

O que é um task queue?

Task queues são usadas como mecanismo de distribuição de trabalho entre threads e máquinas.

Uma entrada de uma fila de tarefa é considerada como uma unidade de trabalho e aqui chamada de Task. Workers dedicados monitoram constantemente essas filas verificando a existência de uma nova tarefa para ser executada.

A comunicação do Celery é realizada via mensagens utilizando um broker para realizar a intermediação entre o client e o worker. Para essa comunicação acontecer, o client insere uma mensagem a fila e o broker realiza o trabalho de distribuir essa mensagem a um worker disponível.

Celery lhe permite trabalhar com múltiplos workers e brokers, facilitando assim uma escalabilidade horizontal na execução de suas tasks.

Requisitos básicos para rodar aplicação utilizando o Celery

Celery necessita de um sistema de mensageria para realizar envios e recebimentos de suas mensagens, sendo os mais utilizados:

• RabbitMQ
• Redis
• Amazon SQS

e para testes locais também aceitando o SQLite.

Entendendo isso, como aplicar?

Aplicando o Celery em uma aplicação Django

Considerando que você já possua uma aplicação rodando com Django, explicarei como instalar o Celery e como mudar aquela task longa para ser executada por um worker em background.

Primeiro passo, escolher um broker:

Neste artigo, utilizaremos o redis devido à sua rápida implementação e ser possível rodar tanto local quanto em produção.

• Instalando no Linux: clique aqui
• Instalando no Windows, clique aqui
• Instalando no Mac OS, clique aqui

E caso se interesse em saber mais sobre o redis, a documentação é bem extensa e rica, e você pode visualizá-la por aqui: redis.io documentation

Para começar a utilizar o Celery em seu projeto é necessário a instalação dele, sendo feito pelo pypy:

$ pip install celery

Após a instalação, criaremos um arquivo chamado celery.py no diretório principal da aplicação, com as seguintes informações:

import os
from celery import Celery
from django.conf import settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings.dir') #1
app = Celery('current_app_name') #2
app.config_from_object('django.conf:settings') #3
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) #4

Fazendo isso, você está:

1. setando em seu environment o modulo referente a settings do Django e caso não exista ele seta o segundo parâmetro informado: settings.dir (informar o modulo onde se encontra sua settings)
2. instanciando o Celery em sua aplicação, informando em qual modulo está o arquivo celery.py
3. está dizendo aonde ele irá encontrar os parametros de configuração do Celery (no nosso caso, estará na settings)
4. ensinando ao Celery aonde ele irá procurar pelos seus arquivos de tasks que em nosso caso ele varrerá todos os apps instalados e buscará arquivos com prefixo tasks

Feito isso, agora precisamos importar o app criado no celery.py em nosso __init__.py na raiz do projeto:

from .celery import app

Isso faz com que ao inicializar a aplicação, inicie também o Celery.

Criado o arquivo principal, iremos configurar o Celery em nossa settings, e, para isso, primeiro vamos entender alguns parâmetros importantes:

Entendendo Queues, Exchanges e Routing Keys

Este é um passo importante para absorver o funcionamento do Celery, pois os passos descritos abaixo representam seu funcionamento, e segue a seguinte sequência:

1. Os clients enviam as mensagens para o Exchange (permutador).
2. O Exchange direciona a mensagem para uma ou mais filas e isso depende muito do tipo de permutação que você está implementando, tendo diversas formas de roteamento e podendo ter diferentes cenários de mensagens.
3. A mensagem aguarda na fila até algum worker disponível a consuma.
4. A mensagem é deletada da fila após um worker confirmar a execução dela.

Celery flow

E para criar o processo de envio e recebimento de mensagens, será necessário:

1. Criar um permutador
2. Criar uma fila
3. Vincular uma fila em um permutador

Tudo isso pode parecer muito trabalhoso, mas na prática é muito simples, basta realizar a seguinte configuração em sua settings.py

from kombu import Exchange, Queue

task_default_queue = 'default' #1
default_exchange = Exchange('media', type='direct') #2
task_queues = (
    Queue(
        'media_queue', #3
        exchange=default_exchange, #4
        routing_key='video' #5
    )
)

1. Configura uma fila default. Caso você não informe a uma task qual fila ela será executada, ela irá buscar neste parâmetro a fila padrão.
2. Cria seu permutador, dando um nome a ele e utilizando seu tipo default direct. Existe outros types, mas é um assunto mais longo e cabe em outro artigo com melhores detalhes.
3. Nome dado a queue, esse nome será o utilizado na hora de informarmos para a task em qual fila ela irá ser executada.
4. Informa a Queue qual Exchange ela irá utilizar
5. Nome da fila criada no sistema de mensageria, onde será postado pelo client e consumido pelo worker.

Criado as rotas de suas filas iremos configurar o broker em nossa aplicação, e para isso, basta adicionar o parâmetro broker_urlno seguinte formato:

broker_url='transport://user:password@hostname:port/virtual_host'

e como estamos rodando o redis em nossa maquina local basta informar:

broker_url='redis://'

Isso já é o suficiente para que você consiga rodar o Celery em sua aplicação, agora iremos alterar aquele sua função super demorada para que ela seja executada em background, tirando o block de sua requisição. Pensando nisso irei criar uma task demorada para exemplificar:

import time

def slow_task():
    print('Started task, processing...')
    time.sleep(120)
    print('Finished Task')
    return True

slow_task()

Essa função é o suficiente para gerar um timeout em sua aplicação, e para transformar ela em uma Celery task, basta você realizar a seguinte alteração:

import time
from celery import shared_task
@shared_task(queue='default') #1
def slow_task():
    print('Started task, processing...')
    time.sleep(120)
    print('Finished Task')
    return True
slow_task.delay() #2

1. Informo que a função decorada será uma Celery task compartilhada, ou seja, ao decorar a função com @shared_task ele retorna um proxy que sempre utiliza a instancia da task no app atual, fazendo ela ser reutilizada independente de quantas instancias de Celery você esteja criando em sua aplicação.
2. Ao invés de executar a função diretamente, agora digo pra ela ser executada de forma assíncrona. Quando executo um .delay() na verdade estou enviando uma mensagem para a fila default para que um worker execute através dessa mensagem a função decorada, fazendo assim sua requisição a esta função não ficar bloqueada aguardando sua finalização.

Rodando os workers do celery

O processo para a execução do Celery é bem simples, basta realizar a seguinte chamada:

celery -A current_app_name worker --loglevel=info

Onde:

• current_app_name: é o nome do modulo onde se encontra o arquivo celery.py, ou seja, o modulo raiz do projeto
• worker: é o parâmetro informando que você está iniciando um nó de worker
• - -loglevel=info: é o parâmetro informando qual o menor level de log que o celery deve logar

Tendo isso como base, você já consegue explorar melhor alguns pontos de sua aplicação aonde os delays são maiores e transformá-los em tasks assíncronas e deixar para que o worker dê conta delas sem impedir a navegação ou uma requisição de seus usuários, ou nos piores casos, retornar um belo timeout para eles. Em um próximo artigo, irei explicar como criar um workflow mais complexo utilizando o Celery Canvas.