Integra Flask y Celery

Flask

Flask es un framework ligero para aplicaciones web con configuración integrada y convenciones predeterminadas que proporcionan consistencia a los desarrolladores en todos los proyectos. Para más información, consulte Página web de Flask.

Apio

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

Requisitos previos

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

• Un clúster de MongoDB. Recomendamos usar Atlas. Para aprender a crear un clúster de Atlas, consulte la página "Comenzar con Atlas" en la documentación de Atlas.

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

• RabbitMQ para utilizar como agente 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

Configurar su aplicación

El archivo config.py contiene configuraciones y credenciales para realizar las siguientes acciones:

• Conecte Celery a RabbitMQ como su agente de mensajes

• Configurar 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 tu correo electrónicoMAIL_USERNAME (, MAIL_PASSWORD MAIL_DEFAULT_SENDERy) para que tu aplicación pueda enviar correos. Por seguridad, te recomendamos generar una contraseña para la aplicación, en lugar de usar tu contraseña principal. Para obtener más información, consulta la configuración de la contraseña de la aplicación en tu cuenta de Google.

También debe proporcionar una cadena de conexión para configurarla como MONGO_URI variable de entorno. Para obtener más información, consulte 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 la Send Newsletter página. El resto de las variables configuran los componentes Flask y Celery.

Inicializar Flask, MongoDB y Celery

El archivo app.py inicializa y configura los componentes principales de la 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 MongoDB newsletter mediante el controlador PyMongo

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

Inicialice Flask, MongoDB y Celery agregando el siguiente código a su 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 apio

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 @celery.task() decorador registra la función como una tarea de Celery. Si bind=True se establece en, la función recibe la instancia de la tarea como self argumento, lo que le permite acceder a los métodos y metadatos de la tarea de Celery. Para obtener más información sobre las tareas, consulte la documentación de la API de 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 URL a una función específica. En el siguiente código, se utiliza para definir las rutas raíz (/), /admin, /subscribe y /send-newsletters. El parámetro opcional methods se utiliza 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

Puede agregar más protecciones de seguridad o personalizar alertas para el usuario de su aplicación en este archivo.

Crea tus plantillas de página

Los archivos HTML del templates directorio definen la interfaz de usuario y están escritos en HTML estándar. Dado que esta aplicación utiliza solicitudes HTTP asíncronas, los scripts de estos archivos utilizan llamadas a la API Fetch. 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

Puede aplicar una hoja de estilo a sus plantillas agregando 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;
}

Pruebe la aplicación

Después de completar los pasos anteriores, tendrá una aplicación funcional que utiliza MongoDB, Flask y Celery para administrar una plataforma de boletín informativo.

Puede utilizar los siguientes pasos para probar su 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:

• Agregue reintentos a su send_emails función

• Dar formato a su boletín informativo

• Implementar funciones de seguridad más rigurosas