# Flask + Celery + Redis 异步发送邮件
### 页面
```python
Flask + Celery Examples
Flask + Celery Examples
Example 1: Send Asynchronous Email
```
### 代码
```python
import os
import random
import time
from flask import Flask, request, render_template, session, flash, redirect, \
url_for, jsonify
from flask_mail import Mail, Message
from celery import Celery
# windows 启动命令 ：celery -A flask_celery:celery worker -l info -P eventlet
app = Flask(__name__)
app.config['SECRET_KEY'] = 'top-secret!'
# Flask-Mail configuration
app.config['MAIL_DEBUG'] = True # 开启debug，便于调试看信息
app.config['MAIL_SUPPRESS_SEND'] = False # 发送邮件，为True则不发送
app.config['MAIL_SERVER'] = 'smtp.qq.com' # 邮箱服务器
app.config['MAIL_PORT'] = 465 # 端口
app.config['MAIL_USE_SSL'] = True # 重要，qq邮箱需要使用SSL
app.config['MAIL_USE_TLS'] = False # 不需要使用TLS
app.config['MAIL_USERNAME'] = '' # 填邮箱
app.config['MAIL_PASSWORD'] = '' # 填授权码
app.config['MAIL_DEFAULT_SENDER'] = '' # 填邮箱，默认发送者
# Celery configuration
app.config['CELERY_BROKER_URL'] = 'redis://@182.92.111.11:6379/1'
app.config['CELERY_RESULT_BACKEND'] = 'redis://@182.92.111.11:6379/2'
# Initialize extensions
mail = Mail(app)
# Initialize Celery
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
@celery.task
def send_async_email(email_data):
"""Background task to send an email with Flask-Mail."""
msg = Message(email_data['subject'],
sender=app.config['MAIL_DEFAULT_SENDER'],
recipients=[email_data['to']])
msg.body = email_data['body']
with app.app_context():
mail.send(msg)
@app.route('/', methods=['GET', 'POST'])
def index():
if request.method == 'GET':
return render_template('falsk_email.html', email=session.get('email', ''))
email = request.form['email']
session['email'] = email
# send the email
email_data = {
'subject': 'Hello from Flask',
'to': email,
'body': 'This is a test email sent from a background Celery task.'
}
if request.form['submit'] == 'Send':
# send right away
send_async_email.delay(email_data)
flash('Sending email to {0}'.format(email))
else:
# send in one minute
send_async_email.apply_async(args=[email_data], countdown=120)
flash('An email will be sent to {0} in one minute'.format(email))
return redirect(url_for('index'))
if __name__ == '__main__':
app.run(debug=True)
```
---
# Agent Instructions: Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.
Perform an HTTP GET request on the current page URL with the `ask` query parameter:
```
GET https://close.gitbook.io/yun-wei-bi-ji/python/flask/flask-+-celery-+-redis-yi-bu-fa-song-you-jian.md?ask=
```
The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.
Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.