flask使用celery傳送郵件

執行本檔案

linux上 # celery -a tasks.celery worker -l info

window上 # celery -a tasks.celery worker --pool=solo -l info

在專案資料夾下建立tasks.py檔案

from flask import flask

from config import config
from celery import celery
from flask_mail import message
from flask_mail import mail
mail = mail(
)#在配置檔案需要配置celery_result_backend、celery_broker_ur
defmake_celery
:'result_backend'],
'celery_broker_url'])
taskbase = celery.task
class
contexttask
(taskbase)
: abstract =
true
def__call__
(self,
*args,
**kwargs):)
:return taskbase.__call__(self,
*args,
**kwargs)
celery.task = contexttask
return celery
@celery.task
defsend_mail
(subject,sender,recipients,body)
: message = message(subject=subject,sender=sender,recipients=recipients,body=body)
mail.send(message)

sqlalchemy_track_modifications =

false
mail_server = os.environ.get(
'mail_server'
) mail_port =
int(os.environ.get(
'mail_port')or
25)mail_use_tls = os.environ.get(
'mail_use_tls')is
notnone
mail_username = os.environ.get(
'mail_username'
) mail_password = os.environ.get(
'mail_password'
)# admins = ['[email protected]']
result_backend =
"redis:"
celery_broker_url =
"redis:"

def

send_password_reset_email
(user)
: tasks.send_mail.delay(
'sad'
'mail_username'],
['[email protected]'],
'sdasd'
)

celery -a tasks.celery worker --pool=solo -l info

然後啟動專案即可 可以試試

使用celery非同步傳送簡訊

1.1 在 celery task mian.py 中新增傳送簡訊函式 celery專案中的所有導包位址,都是以celery base dir為基準設定.執行celery命令時,也需要進入celery base dir目錄執行.celery base dir os.path.dirname os.p...

Flask 使用Celery並且使其接入上下文

tips 使用工廠函式接入 code celery配置檔案 celery config.py broker url redis result backend redis task serializer json result serializer json accept content json t...

非同步傳送郵件 celery

包結構 依賴包 redis3.4.1 celery4.4.1 eventlet 0.25.1 首先建立乙個celery tasks的python包,在celery tasks的 init py中配置 drugstore是你專案的主目錄名稱，即有settings.py檔案的包名 這個需要改為你自己的 ...