Integra Flask y Celery

Flask

Flask es un framework de aplicaciones web liviano con configuración incorporada y convenciones predeterminadas que brindan coherencia a los desarrolladores en todos los proyectos. Para más información, consulta la Página web de Flask.

Celery

Celery es una cola de tareas distribuida de código abierto que gestiona grandes volúmenes de mensajes de forma eficiente. Permite el procesamiento asíncrono y la programación de tareas. Para más información, consulta la página web de Celery.

Requisitos previos

Asegúrese de tener los siguientes componentes instalados y configurados antes de iniciar este tutorial:

• Un clúster de MongoDB. Recomendamos usar Atlas. Para aprender a crear un clúster de Atlas, consulte la guía de inicio rápido de Atlas.

• Una base de datos llamada newsletter en tu clúster. Para obtener más información, consulta la página Crear una base de datos en la guía de Atlas.

• RabbitMQ para usarlo como broker de mensajes para Celery.

• Gmail como servidor SMTP. Para más información sobre servidores SMTP, consulta la página de Wikipedia sobre el Protocolo Simple de Transferencia de Correo.

• Python 3.9 o posterior

Configuración

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

Configura tu aplicación

El archivo config.py contiene la configuración y las credenciales para realizar las siguientes acciones:

• Conectar Celery a RabbitMQ como su intermediario de mensajes

• Configura Flask-Mail para utilizar Gmail como su servidor SMTP.

• Conecte su aplicación a su implementación de MongoDB

Defina las configuraciones necesarias agregando el siguiente código a su archivo 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'

Debes proporcionar tus credenciales de Gmail y correo electrónico (MAIL_USERNAME, MAIL_PASSWORD y MAIL_DEFAULT_SENDER) para habilitar que tu aplicación envíe correos electrónicos. Por motivos de seguridad, recomendamos generar una contraseña de aplicación para usar, en lugar de utilizar la contraseña primaria. Para más información, consulta la Configuración de la contraseña de la aplicación en tu cuenta de Google.

También debes proporcionar una cadena de conexión para establecerla como la variable de entorno MONGO_URI. Para obtener más información, consulta la sección Crear una cadena de conexión de esta guía.

La URLCELERY_BROKER_URL del broker de Celery () especifica RabbitMQ como su broker, pero puede personalizarla para que sea compatible con otras implementaciones. Para más información, consulte la sección "Configuración del broker" de la documentación de Celery.

La lista ALLOWED_IPS se utiliza para controlar el acceso a Send Newsletter página. El resto de las variables configuran los componentes de Flask y Celery.

Inicializar Flask, MongoDB y Celery

El archivo app.py inicializa y configura los componentes principales de su aplicación. Realiza las siguientes tareas:

• Crea una aplicación Flask y carga constantes de configuración

• Inicializa una instancia de Flask-Mail con la configuración del servidor de correo de la aplicación.

• Se conecta a la base de datos newsletter MongoDB utilizando el controlador PyMongo

• Crea una instancia de Celery configurada con la aplicación Flask y el agente elegido

Inicializa Flask, MongoDB y Celery añadiendo el siguiente código a tu archivo 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)

Crear una tarea de Celery

La tarea Celery utiliza los componentes instanciados en su archivo app.py para enviar un boletín informativo por correo electrónico a los suscriptores.

El decorador @celery.task() registra la función como una tarea de Celery. Configurar bind=True significa que la función recibe la instancia de la tarea como el argumento self, lo que le permite acceder a los métodos y metadatos de la tarea Celery. Para obtener más información sobre las tareas, consulta la documentación de la API celery.app.task.

Dado que esta tarea se ejecuta fuera del ciclo de solicitud HTTP de Flask, debe proporcionar manualmente el contexto de la aplicación envolviendo la lógica del correo electrónico en un bloque with app.app_context(). Esto permite a Flask acceder a otros componentes, como la instancia Flask-Mail mail y la conexión de PyMongo a su base de datos MongoDB newsletter.

Esta función recorre la lista de subscribers, crea un correo electrónico con la clase Flask-Mail Message y lo envía a cada usuario mediante el objeto mail. Tras el envío de cada correo electrónico, registra la entrega insertando un documento en la colección deliveries de MongoDB para registrar el envío del mensaje. Cada operación de correo electrónico se encapsula en un bloque try para garantizar que, en caso de error, se registre el fallo y la base de datos no se actualice con un registro de entrega falso.

Define tu función send_emails() agregando el siguiente código a tu archivo 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'}

Define tus rutas

En Flask, el decorador @app.route() asigna una ruta de URL a una función específica. En el siguiente código, se utiliza para definir la raíz (/), /admin, /subscribe y /send-newsletters rutas. El parámetro opcional methods se usa en algunos casos para definir una lista de métodos HTTP permitidos.

El decorador @app.before_request() establece una función que se ejecuta antes de cada solicitud. En este caso, la función proporciona seguridad básica al limitar el acceso a la página admin a las direcciones IP indicadas en el parámetro ALLOWED_IPS definido en el archivo config.py. En concreto, solo se permite el acceso a la página localhost.

Las rutas raíz y /admin representan páginas mediante el método render_template(). Las rutas /subscribe y /send-newsletters acceden a los parámetros de solicitud en request.form[] para ejecutar comandos y luego devuelven respuestas HTTP.

Define tus rutas agregando el siguiente código a tu archivo 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

Puedes añadir más protecciones de seguridad o personalizar las alertas orientadas a los usuarios para tu aplicación en este archivo.

Crea tus plantillas de página

Los archivos HTML en el directorio templates definen la interfaz de usuario y están escritos utilizando HTML estándar. Debido a que esta aplicación utiliza solicitudes HTTP asíncronas, los scripts de estos archivos utilizan Llamadas Fetch API. Estos scripts también gestionan tiempos de espera y errores.

<!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>

Formatear sus páginas

Puedes aplicar una hoja de estilos a tus plantillas añadiendo el siguiente código al archivo 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;
}

Prueba la aplicación

Después de completar los pasos anteriores, tendrás una aplicación funcional que utiliza MongoDB, Flask y Celery para gestionar una plataforma de boletines informativos.

Puedes usar los siguientes pasos para probar tu aplicación:

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 pasos

Esta aplicación demuestra cómo integrar una aplicación Flask con la cola de tareas de Celery para gestionar los datos de los suscriptores y enviar correos electrónicos por lotes. Puede desarrollar esta aplicación para experimentar con Flask o Celery. Algunas posibles mejoras incluyen los siguientes cambios:

• Añade retries a tu función send_emails

• Formatear tu boletín

• Implementa funcionalidades de seguridad más rigurosas funcionalidades de seguridad