Tutorial: Integração de balão e aipo

Flask

O Spark é uma estrutura leve de aplicação da Web com configurações integradas e padrões de convenções que fornecem consistência aos desenvolvedores em todos os projetos. Para obter mais informações, consulte a página da Web do Flask.

Aipo

O Cely é uma fila de tarefas distribuída de código aberto que lida com grandes volumes de mensagens de forma eficiente. Suporta processamento assíncrono e agendamento de tarefas. Para obter mais informações, consulte a página da Web do Aipo.

Pré-requisitos

Certifique-se de ter os seguintes componentes instalados e configurados antes de iniciar este tutorial:

• Um cluster MongoDB . Recomendamos que você use o Atlas. Para saber como criar um cluster do Atlas , consulte a página Iniciar com Atlas na documentação do Atlas .

• Um banco de dados denominado newsletter em seu cluster. Para mais informações, consulte a página Criar um Banco de Dados no guia do Atlas .

• deverá ser usado como Corretor de Mensagens para o Aipo.

• Gmail para usar como servidor SMTP . Para obter mais informações sobre servidores SMTP, consulte a página da Wikipedia sobre o Protocolo de Transferência de Correio Simples.

• Python 3.9 ou posterior

Configurar

mkdir newsletter
cd newsletter

newsletter/
├── app.py
├── config.py
├── routes.py
├── tasks.py
├── templates/
│   ├── admin.html
│   └── subscribe.html
└── static/
    └── styles.css

pip install flask-pymongo Flask-Mail celery

Configurar seu aplicativo

O arquivo config.py contém configurações e credenciais para executar as seguintes ações:

• Conecte o Aipo ao RubyMQ como seu corretor de mensagens

• Configurar oFlask-Mail para usar o Gmail como servidor SMTP

• Conecte seu aplicação ao deployment do MongoDB

Defina as configurações necessárias adicionando o seguinte código ao seu arquivo config.py:

import os
class Config:
   MAIL_USERNAME = '<username>' # Your email address without '@gmail.com'
   MAIL_PASSWORD = '<app password>'
   MAIL_DEFAULT_SENDER = '<email address>'
   MONGO_URI = '<mongodb connection string>'
   DATABASE_NAME = "newsletter"
   ALLOWED_IPS = ['127.0.0.1']
   MAIL_SERVER = 'smtp.gmail.com'
   MAIL_PORT = 587
   MAIL_USE_TLS = True
   CELERY_BROKER_URL = 'amqp://guest:guest@localhost//'
   RESULT_BACKEND = MONGO_URI + '/celery_results'

Você deve fornecer suas credenciais e e-mail do GmailMAIL_USERNAME(, MAIL_PASSWORD MAIL_DEFAULT_SENDERe) para permitir que seu aplicação envie e-mails. Por motivos de segurança, recomendamos que você gere uma senha de aplicação para usar, em vez de usar sua senha primária. Para obter mais informações, consulte as configurações de senha do app na sua Conta do Google.

Você também deve fornecer uma string de conexão para definir como a MONGO_URI variável de ambiente do. Para obter mais informações, consulte a seção Criar uma connection string deste guia.

A URL de Corretora de Aipo fornecidaCELERY_BROKER_URL() especifica o RubyMQ como seu Corretor, mas você pode personalizar essa URL para suportar outras implementações. Para obter mais informações, consulte a seção Configurações do Corretor da documentação do Aipo.

A lista ALLOWED_IPS é utilizada para controlar o acesso à página Send Newsletter. O restante das variáveis configura os componentes Pipeline e Aipo.

Inicializar Pipeline, MongoDB e Aipo

O arquivo app.py inicializa e configura os componentes principais do seu aplicação. Ele executa as seguintes tarefas:

• Cria um aplicação Spark e carrega constantes de configuração

• Inicializa uma instância do Flusk-Mail com as configurações do servidor de e-mail do aplicativo

• Conecta-se ao banco de dados MongoDB newsletter usando o driver PyMongo

• Cria uma instância de aipo configurada com a aplicação Sparkle e o agente escolhido

Inicialize Spark, MongoDB e Aipo adicionando o seguinte código ao seu arquivo app.py:

from flask import Flask
from flask_mail import Mail
from flask_pymongo import PyMongo
from celery import Celery
# Create a Flask application
app = Flask(__name__)
app.config.from_object('config.Config')
# Create a Flask-Mail instance
mail = Mail(app)
# Connect to MongoDB
client = PyMongo(app).cx
db = client[app.config['DATABASE_NAME']]
# Create a Celery instance
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
from routes import *
from tasks import *
if __name__ == '__main__':
   app.run(debug=True)

Criar uma Tarefa de Aipo

A tarefa Aipo usa os componentes instanciados em seu arquivo app.py para enviar um e-mail de boletim informativo aos assinantes.

O @celery.task() construtor registra a função como uma tarefa de aipo. Definir bind=True significa que a função recebe a instância de tarefa como self argumento, o que lhe permite acessar os métodos e metadados de tarefa do Aipo. Para obter mais informações sobre tarefas, consulte a documentação da API acelary.app.task.

Como essa tarefa é executada fora do ciclo de solicitação HTTP do Spark, você deve fornecer manualmente o contexto do aplicação encapsulando a lógica de e-mail em um bloco with app.app_context(). Isso dá ao Pipeline acesso a outros componentes, como a instância do Pipeline mail e a conexão do PyMongo com seu banco de dados newsletter MongoDB .

Esta função percorre a lista de subscribers, cria um e-mail utilizando a classe Frasco-Mail Message e, em seguida, envia para cada usuário utilizando o objeto mail. Após cada e-mail ser enviado, ele registra a entrega inserindo um documento em sua coleção MongoDB deliveries para registrar que a mensagem foi enviada. Cada operação de e-mail é envolta em um bloco try para garantir que, no caso de um erro, a falha seja registrada e o banco de dados não seja atualizado com um falso registro de entrega.

Defina sua função send_emails() adicionando o seguinte código ao seu arquivo tasks.py:

from flask_mail import Message
from app import app, mail, db, celery
from datetime import datetime
@celery.task(bind=True)
def send_emails(self, subscribers, title, body):
   with app.app_context():
      for subscriber in subscribers:
            try:
               print(f"Sending email to {subscriber['email']}")
               msg = Message(title, recipients=[subscriber['email']])
               msg.body = body
               mail.send(msg)
               db.deliveries.insert_one({
                  'email': subscriber['email'],
                  'title': title,
                  'body': body,
                  'delivered_at': datetime.utcnow()
               })
               print("Email sent")
            except Exception as e:
               print(f"Failed to send email to {subscriber['email']}: {str(e)}")
      return {'result': 'All emails sent'}

Defina suas rotas

No Piper, o construtor do @app.route() atribui um caminho de URL para uma função específica. No código seguinte, é utilizado para definir as rotas raiz (/), /admin, /subscribe e /send-newsletters. O parâmetro opcional methods é utilizado em algumas instâncias para definir uma lista de métodos HTTP permitidos.

O planejador @app.before_request() define uma função para ser executada antes de cada solicitação. Nesse caso, a função fornece alguma segurança básica limitando o acesso à página admin aos endereços IP listados no parâmetro ALLOWED_IPS definido no arquivo config.py. Especificamente, o acesso só é permitido para o localhost.

As rotas raiz e /admin renderizam páginas utilizando o método render_template(). As rotas /subscribe e /send-newsletters acessam os parâmetros da solicitação em request.form[] para executar comandos e, em seguida, retornar respostas HTTP.

Defina suas rotas adicionando o seguinte código ao seu arquivo routes.py:

from flask import render_template, request, abort, jsonify
from app import app, db
from tasks import send_emails
@app.before_request
def limit_remote_addr():
   if 'X-Forwarded-For' in request.headers:
      remote_addr = request.headers['X-Forwarded-For'].split(',')[0]
   else:
      remote_addr = request.remote_addr
   if request.endpoint == 'admin' and remote_addr not in app.config['ALLOWED_IPS']:
      abort(403)
@app.route('/')
def home():
   return render_template('subscribe.html')
@app.route('/admin')
def admin():
   return render_template('admin.html')
@app.route('/subscribe', methods=['POST'])
def subscribe():
   first_name = request.form['firstname']
   last_name = request.form['lastname']
   email = request.form['email']
   if db.users.find_one({'email': email}):
      return """
      <div class="response error">
            <span class="icon">&#x2716;</span> This email is already subscribed!
      </div>
      """, 409
   db.users.insert_one({'firstname': first_name, 'lastname': last_name, 'email': email, 'subscribed': True})
   return """
   <div class="response success">
      <span class="icon">&#x2714;</span> Subscribed successfully!
   </div>
   """, 200
@app.route('/send-newsletters', methods=['POST'])
def send_newsletters():
   title = request.form['title']
   body = request.form['body']
   subscribers = list(db.users.find({'subscribed': True}))
   for subscriber in subscribers:
      subscriber['_id'] = str(subscriber['_id'])
   send_emails.apply_async(args=[subscribers, title, body])
   return jsonify({'message': 'Emails are being sent!'}), 202

Você pode adicionar mais proteções de segurança ou personalizar alertas direcionados ao usuário para seu aplicação neste arquivo.

Crie seus modelos de página

Os arquivos HTML no templates diretório definem a interface do usuário e são gravados usando HTML padrão. Como esse aplicação usa solicitações HTTP assíncronas, os scripts nesses arquivos usam chamadas de API de busca. Esses scripts também lidam com tempos limite e erros.

<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <meta name="viewport" content="width=device-width, initial-scale=1.0">
   <title>Subscribe to Newsletter</title>
   <link rel="stylesheet" href="{{ url_for('static', filename='styles.css') }}">
</head>
<body>
   <h1>Subscribe to our Newsletter</h1>
   <form id="subscribe-form">
      <label for="firstname">First Name:</label>
      <input type="text" id="firstname" name="firstname" required>
      <br>
      <label for="lastname">Last Name:</label>
      <input type="text" id="lastname" name="lastname" required>
      <br>
      <label for="email">Email:</label>
      <input type="email" id="email" name="email" required>
      <br>
      <button type="submit">Subscribe</button>
   </form>
   <div id="response"></div>
   <script>
      document.getElementById('subscribe-form').addEventListener('submit', function(event) {
            event.preventDefault();
            var formData = new FormData(event.target);
            fetch('/subscribe', {
               method: 'POST',
               body: formData
            }).then(response => {
               if (!response.ok) {
                  throw response;
               }
               return response.text();
            }).then(data => {
               document.getElementById('response').innerHTML = data;
               document.getElementById('subscribe-form').reset();
               setTimeout(() => {
                  document.getElementById('response').innerHTML = '';
               }, 3000);
            }).catch(error => {
               error.text().then(errorMessage => {
                  document.getElementById('response').innerHTML = errorMessage;
                  setTimeout(() => {
                        document.getElementById('response').innerHTML = '';
                  }, 3000);
               });
            });
      });
   </script>
</body>
</html>

<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <meta name="viewport" content="width=device-width, initial-scale=1.0">
   <title>Admin - Send Newsletter</title>
   <link rel="stylesheet" href="{{ url_for('static', filename='styles.css') }}">
</head>
<body>
   <h1>Send Newsletter</h1>
   <form id="admin-form">
      <label for="title">Title:</label>
      <input type="text" id="title" name="title" required>
      <br>
      <label for="body">Body:</label>
      <textarea id="body" name="body" required></textarea>
      <br>
      <button type="submit">Send</button>
   </form>
   <div id="response"></div>
   <script>
      document.getElementById('admin-form').addEventListener('submit', function(event) {
            event.preventDefault();
            var formData = new FormData(event.target);
            fetch('/send-newsletters', {
               method: 'POST',
               body: formData
            })
            .then(response => response.json())
            .then(() => {
               document.getElementById('response').innerText = 'Emails are being sent!';
               setTimeout(() => {
                  document.getElementById('response').innerText = '';
               }, 3000);
               document.getElementById('admin-form').reset();
            })
            .catch(error => {
               document.getElementById('response').innerText = 'Error sending emails.';
               setTimeout(() => {
                  document.getElementById('response').innerText = '';
               }, 3000);
               console.error('Error:', error);
            });
      });
   </script>
</body>
</html>

Formatar Suas Páginas

Você pode aplicar uma folha de estilo aos seus modelos adicionando o seguinte código ao arquivo styles.css:

body {
   font-family: system-ui;
   font-optical-sizing: auto;
   font-weight: 300;
   font-style: normal;
   margin: 0;
   padding: 0;
   display: flex;
   flex-direction: column;
   align-items: center;
   justify-content: center;
   height: 100vh;
   background-color: #040100;
}
h1 {
   color: white;
}
form {
   background: #023430;
   padding: 30px 40px;
   border-radius: 8px;
   box-shadow: 0 0 10px rgba(0, 0, 0, 0.1);
   width: 100%;
   max-width: 400px;
   margin: 20px 0;
}
label {
   display: block;
   margin-bottom: 8px;
   font-weight: bold;
   color: white;
}
input[type="text"],
input[type="email"],
textarea {
   width: 100%;
   padding: 10px;
   margin-bottom: 10px;
   border: 1px solid #ccc;
   border-radius: 4px;
   font-size: 16px;
}
button {
   background: #00ED64;
   color: white;
   padding: 10px 20px;
   border: none;
   border-radius: 4px;
   cursor: pointer;
   font-size: 16px;
   font-family: "Nunito", sans-serif;
}
button:hover {
   background: #00684A;
}
#response {
   margin-top: 20px;
   font-size: 16px;
   color: #28a745;
}
footer {
   text-align: center;
   padding: 20px;
   margin-top: 20px;
   font-size: 16px;
   color: #666;
}

Teste o Aplicativo

Depois de concluir as etapas anteriores, você tem um aplicação funcional que usa MongoDB, Piper e Aipo para gerenciar uma plataforma de boletim informativo.

Você pode usar as seguintes etapas para testar seu aplicação:

brew services start rabbitmq

rabbitmq-service start

sudo systemctl start rabbitmq-server

flask --app app run

celery -A app.celery worker --loglevel=info

[2025-05-27 09:54:43,873: INFO/ForkPoolWorker-7] Task tasks.send_emails[7d7f9616-7b9b-4508-a889-95c35f54fe43] succeeded in 3.93334774998948s: {'result': 'All emails sent'}
[2025-05-27 10:04:52,043: INFO/MainProcess] Task tasks.send_emails[ac2ec70f-2d3e-444a-95bb-185ac659f460] received
[2025-05-27 10:04:52,046: WARNING/ForkPoolWorker-7] Sending email to <subscriber_email>
[2025-05-27 10:04:53,474: WARNING/ForkPoolWorker-7] Email sent

Próximos passos

Este aplicação demonstra como integrar um aplicação Pipeline com a fila de tarefas do Aipo para gerenciar dados de assinantes e enviar e-mails em lote . Você pode construir este aplicação para experimentar com Frasco ou Aipo. Algumas possíveis melhorias incluem as seguintes alterações:

• Adicionar novas tentativas à sua send_emails função

• Formatar sua newsletter

• Implemente funcionalidades de segurança mais rigorosas