import random
from flask import Flask, request, jsonify, render_template, current_app
from celery import Celery
from flask_mail import Message, Mail
import settings
app = Flask(__name__)
app.config.from_object(settings.BaseConfig)
mail = Mail(app)
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
@celery.task
def send_async_email(subject, to, expire_minute):
app = current_app._get_current_object()
msg = Message(subject=subject,
sender=app.config['MAIL_DEFAULT_SENDER'],
recipients=[to])
msg.body = 'This is a test email sent from a background Celery task.'
vcode = ''.join([str(i) for i in random.sample(range(0, 10), 4)])
context = {
'subject': subject,
'expire_minute': expire_minute,
'vcode': vcode,
# 'debug': get_config_partial('DEBUG'),
# 'debug': True,
}
msg.html = render_template(app.config['EMAIL_TEMPLATE'] + '.html', **context)
# with app.app_context():
mail.send(msg)
@app.route('/send_email', methods=['POST'])
def send_email():
try:
subject = request.form.get("subject")
to = request.form.get("to")
expire_minute = request.form.get("expire_minute")
except:
return jsonify({'msg': '参数有误'})
# send_async_email.delay(app, msg)
send_async_email.apply_async(args=[subject, to, expire_minute])
message = '邮件发送成功'
return jsonify(message=message)
if __name__ == '__main__':
app.run()
这是我的settings文件
class BaseConfig(object):
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'
BACKEND = None
RABBITMQ_QUEUE = 'hello'
MAIL_SERVER = 'smtp.163.com '
MAIL_PORT = 25
MAIL_USE_TLS = True
MAIL_USERNAME = 'mini@ 163.com'
MAIL_PASSWORD = '1314AC'
MAIL_DEFAULT_SENDER = 'mini @ 163.com'
DEFAULT_FROM_EMAIL = 'mini @ 163.com'
EMAIL_TEMPLATE = 'email'