0.配置文件

# coding:utf-8BROKER_URL = "redis://127.0.0.1:6379/1" CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/2'

1.启动文件

# coding:utf-8from celery import Celery from ihome.tasks import config# 定义celery对象 celery_app = Celery("ihome")# 引入配置信息 celery_app.config_from_object(config)# 自动搜寻异步任务 celery_app.autodiscover_tasks(["ihome.tasks.sms"])

2.发送短信辅助类

# coding=utf-8from CCPRestSDK import REST# 主帐号 accountSid = '8aaf0708568d4143015697b0f4960888'# 主帐号Token accountToken = '42d3191f0e6745d6a9ddc6c795da0bed'# 应用Id appId = '8aaf0708568d4143015697b0f56e088f'# 请求地址，格式如下，不需要写http:// serverIP = 'app.cloopen.com'# 请求端口 serverPort = '8883'# REST版本号 softVersion = '2013-12-26'# 发送模板短信# @param to 手机号码# @param datas 内容数据 格式为列表 例如：['12','34']，如不需替换请填 ''# @param $tempId 模板Idclass CCP(object):"""自己封装的发送短信的辅助类"""# 用来保存对象的类属性instance = Nonedef __new__(cls):# 判断CCP类有没有已经创建好的对象，如果没有，创建一个对象，并且保存# 如果有，则将保存的对象直接返回if cls.instance is None:obj = super(CCP, cls).__new__(cls)# 初始化REST SDKobj.rest = REST(serverIP, serverPort, softVersion)obj.rest.setAccount(accountSid, accountToken)obj.rest.setAppId(appId)cls.instance = objreturn cls.instancedef send_template_sms(self, to, datas, temp_id):""""""result = self.rest.sendTemplateSMS(to, datas, temp_id)# for k, v in result.iteritems():## if k == 'templateSMS':# for k, s in v.iteritems():# print '%s:%s' % (k, s)# else:# print '%s:%s' % (k, v)# smsMessageSid:ff75e0f84f05445ba08efdd0787ad7d0# dateCreated:20171125124726# statusCode:000000status_code = result.get("statusCode")if status_code == "000000":# 表示发送短信成功return 0else:# 发送失败return -1

3.发送短信后端逻辑

# GET /api/v1.0/sms_codes/<mobile>?image_code=xxxx&image_code_id=xxxx @api.route("/sms_codes/<re(r'1[34578]\d{9}'):mobile>") def get_sms_code(mobile):"""获取短信验证码"""# 获取参数image_code = request.args.get("image_code")image_code_id = request.args.get("image_code_id")# 校验参数if not all([image_code_id, image_code]):# 表示参数不完整return jsonify(errno=RET.PARAMERR, errmsg="参数不完整")# 业务逻辑处理# 从redis中取出真实的图片验证码try:real_image_code = redis_store.get("image_code_%s" % image_code_id)except Exception as e:current_app.logger.error(e)return jsonify(errno=RET.DBERR, errmsg="redis数据库异常")# 判断图片验证码是否过期if real_image_code is None:# 表示图片验证码没有或者过期return jsonify(errno=RET.NODATA, errmsg="图片验证码失效")# 删除redis中的图片验证码，防止用户使用同一个图片验证码验证多次try:redis_store.delete("image_code_%s" % image_code_id)except Exception as e:current_app.logger.error(e)# 与用户填写的值进行对比if real_image_code.lower() != image_code.lower():# 表示用户填写错误return jsonify(errno=RET.DATAERR, errmsg="图片验证码错误")# 判断对于这个手机号的操作，在60秒内有没有之前的记录，如果有，则认为用户操作频繁，不接受处理try:send_flag = redis_store.get("send_sms_code_%s" % mobile)except Exception as e:current_app.logger.error(e)else:if send_flag is not None:# 表示在60秒内之前有过发送的记录return jsonify(errno=RET.REQERR, errmsg="请求过于频繁，请60秒后重试")# 判断手机号是否存在try:user = User.query.filter_by(mobile=mobile).first()except Exception as e:current_app.logger.error(e)else:if user is not None:# 表示手机号已存在return jsonify(errno=RET.DATAEXIST, errmsg="手机号已存在")# 如果手机号不存在，则生成短信验证码sms_code = "%06d" % random.randint(0, 999999)# 保存真实的短信验证码try:redis_store.setex("sms_code_%s" % mobile, constants.SMS_CODE_REDIS_EXPIRES, sms_code)# 保存发送给这个手机号的记录，防止用户在60s内再次出发发送短信的操作redis_store.setex("send_sms_code_%s" % mobile, constants.SEND_SMS_CODE_INTERVAL, 1)except Exception as e:current_app.logger.error(e)return jsonify(errno=RET.DBERR, errmsg="保存短信验证码异常")# 发送短信# 使用celery异步发送短信, delay函数调用后立即返回（非阻塞）#send_sms.delay(mobile, [sms_code, int(constants.SMS_CODE_REDIS_EXPIRES/60)], 1)# 返回异步任务的对象result_obj = send_sms.delay(mobile, [sms_code, int(constants.SMS_CODE_REDIS_EXPIRES/60)], 1)print(result_obj.id)# 通过异步任务对象的get方法获取异步任务的结果, 默认get方法是阻塞的ret = result_obj.get()print("ret=%s" % ret)# 返回值# 发送成功return jsonify(errno=RET.OK, errmsg="发送成功")

4.client相关代码

# coding:utf-8from celery import Celery from ihome.libs.yuntongxun.sms import CCP# 定义celery对象 celery_app = Celery("ihome", broker="redis://127.0.0.1:6379/1")@celery_app.task def send_sms(to, datas, temp_id):"""发送短信的异步任务"""ccp = CCP()ccp.send_template_sms(to, datas, temp_id)# celery开启的命令 # celery -A ihome.tasks.task_sms worker -l info

5.worker相关代码

# coding:utf-8from ihome.tasks.main import celery_app from ihome.libs.yuntongxun.sms import CCP# # @celery_app.task # def send_sms(to, datas, temp_id): # """发送短信的异步任务""" # pass@celery_app.task def send_sms(to, datas, temp_id):"""发送短信的异步任务"""ccp = CCP()try:result = ccp.send_template_sms(to, datas, temp_id)except Exception as e:result = -2return result

总结

以上是生活随笔为你收集整理的Flask爱家租房--celery（发送验证短信）的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。