在Flask中使用Celery完成异步和定时任务（Flask、Celery、Redis）

编程目标

通过使用Flask和Celery，实现一个简单的Web应用程序，能够接收HTTP POST请求，并异步发送电子邮件。

说明

1. 使用Flask创建一个简单的Web应用程序，包含一个HTTP POST路由，用于接收发送电子邮件的请求。
2. 使用Celery实现一个异步任务，用于发送电子邮件。
3. 发送电子邮件的请求应包含以下信息：
   • 收件人地址
   • 邮件主题
   • 邮件内容
4. 邮件发送成功后，返回响应表示成功发送。

技术栈

• Python
• Flask
• Celery
• Redis

接口设计

1. 发送邮件接口

• URL: /send-email
• 方法: POST
• 请求参数:
  • recipient (string): 收件人地址
  • subject (string): 邮件主题
  • body (string): 邮件内容
• 成功响应:
  • 状态码: 202 Accepted
  • 响应体: {“message”: “邮件发送任务已启动”}

app.py代码：

from flask import Flask, request, jsonify
from tasks import send_email_async

app = Flask(__name__)
app.json.ensure_ascii = False  # 解决中文乱码问题
@app.route('/send-email', methods=['POST'])
def email_sender():
    post_form_data = request.json
    print(post_form_data)
    # 调用异步发送邮件任务
    email_data = {
        'sender_email': 'csdn_代码写注释@163.com',
        'sender_password': 'csdn_代码写注释',
        'recipient': post_form_data['recipient'],
        'subject': post_form_data['subject'],
        'body': post_form_data['body']
    }
    send_email_async.delay(email_data)
    return jsonify({"message": "邮件发送任务已启动"}), 202

if __name__ == "__main__":
    app.run(debug=True, host='0.0.0.0', port=5000)

tasks.py代码：

from celery import Celery
import smtplib
from email.mime.text import MIMEText
from email.header import Header

app = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

@app.task
def send_email_async(email_data):
    content = email_data.get('body')
    msg_from = email_data.get('sender_email')
    password = email_data.get('sender_password')
    msg_to = email_data.get('recipient')
    subject = email_data.get('subject')

    msg = MIMEText(content)
    msg['Subject'] = subject
    msg['From'] = msg_from
    msg['To'] = msg_to

    try:
        s = smtplib.SMTP_SSL("smtp.163.com", 465)
        s.login(msg_from, password)
        s.sendmail(msg_from, msg_to, msg.as_string())
        s.quit()
        print('邮件发送成功!')
        return "邮件发送成功!"
    except Exception as e:
        print(f"邮件发送失败: {e}")
        return "邮件发送失败"

test_send_email.py

import requests

subject = "药价监督流水报告"  # 主题
content = """
尊敬的xxx客户，

随函附上本季度药价监督的流水报告。以下是本季度药价监管的关键要点：

1. 监督范围：全国23个省市的主要药品批发市场及在线药品交易平台。
2. 检查次数：共计1,536次现场检查和3,245次在线监控。
3. 发现问题：在检查中发现15起价格违规行为，涉及7种药品。
4. 违规处理：所有违规行为均已记录在案，并对相关企业进行了警告及罚款处理。
5. 价格波动：本季度药品平均价格波动率为3.5%，与上季度相比下降了1.2个百分点。

2024年05月09日
"""
def send_email():
    url = 'http://localhost:5000/send-email'  # Flask 应用的 URL
    data = {
        'recipient': 'csdn_代码写注释@qq.com',  # 收件人地址
        'subject': subject,  # 邮件主题
        'body': content  # 邮件内容
    }

    # 发送 POST 请求
    response = requests.post(url, json=data)

    # 输出响应内容
    print('响应状态码是:', response.status_code)
    print('响应内容是：', response.text)

send_email()

实现效果：


requirements.txt:

amqp==5.2.0
async-timeout==4.0.3
billiard==4.2.0
blinker==1.8.2
celery==5.4.0
certifi==2024.2.2
charset-normalizer==3.3.2
click==8.1.7
click-didyoumean==0.3.1
click-plugins==1.1.1
click-repl==0.3.0
colorama==0.4.6
dnspython==2.6.1
eventlet==0.36.1
Flask==3.0.3
Flask-Mail==0.9.1
greenlet==3.0.3
idna==3.7
itsdangerous==2.2.0
Jinja2==3.1.4
kombu==5.3.7
MarkupSafe==2.1.5
prompt-toolkit==3.0.43
python-dateutil==2.9.0.post0
redis==5.0.4
requests==2.31.0
six==1.16.0
tzdata==2024.1
urllib3==2.2.1
vine==5.1.0
wcwidth==0.2.13
Werkzeug==3.0.3

如何启动程序？

# step:0
新建虚拟环境
# step:1
pip install -r requirements.txt
# step:2
启动Redis服务
# 见第一张图片，分别打开三个窗口
窗口1：app
在这个窗口里运行命令：
python app.py

窗口2：tests
在这个窗口里运行命令：
python test_send_email.py

窗口3：Celery
在这个窗口里运行命令：
celery -A tasks  worker --pool=solo --loglevel=info

然后你把代码里的发送方邮箱、发送方密码（注意是授权码）、接收方邮箱这三个配置参数改成真实的，即可运行本程序实现Celery完成异步发送邮件了。

愿你的努力，被世界看到。