在一些业务系统中，整合短信和邮件通知是一种常见的处理方式，之前我在多篇随笔中介绍过基于.NET的整合处理，本篇随笔介绍基于FastAPI的Python开发框架后端，增加阿里云短信和邮件发送通知处理。
之前我在多篇随笔中介绍过基于.NET的《SqlSugar开发框架》中整合过短信接入的内容：《使用阿里云的短信服务发送短信》、《基于SqlSugar的开发框架循序渐进介绍（17）-- 基于CSRedis实现缓存的处理》、《循序渐进VUE+Element 前端应用开发(32）--- 手机短信动态码登陆处理》。
短信通知，一般用于系统的登录，或者对重要数据变更的身份确认，各个平台都相关的短信接口，不过好像华为服务器已经不提供短信接入了，阿里云还可以，或者也可以找一些其他的短信服务商，基本上都会提供相应的接口或者SDK，对接还是很方便的。
本篇随笔基于阿里云的短信接口进行对接短信通知，也主要就是解决BS端 或者H5端的身份登录及密码修改、重置等重要处理的通知。
对于Python开发来说，基于阿里云的短信处理，可以使用它的 alibabacloud_dysmsapi20170525 SDK包来进行对接，虽然这个包时间上比较老，好像也没有看到更新的SDK了。
使用pip 命令安装了该SDK即可。
pip install alibabacloud_dysmsapi20170525
我们使用阿里云API发送短信，一般需要提供下面的身份信息。
使用阿里云的短信服务，需要注册登录自己的阿里云控制台，然后进入AccessKeys的处理界面
这里我们获取到AccessKey ID 和Access Key Secret两个关键信息，需要用在数据签名的里面的。
发送接口还需要提供下面的的一些信息，包括必要的手机号码，签名，服务器端模板代码，以及短信码等必要信息。
签名，一般为我们短信提示的公司名称，如【广州爱奇迪】这样的字样。
服务器端模板代码，阿里云默认提供了一些基础模板，我们可以从中选择。
如下具体接口请求需要提供的JSON数据。
{ "phone_numbers": "13800138000", "sign_name": "YourSignName", "template_code": "SMS_123456789", "template_param": { "code": "123456" } }
短信真实接收到的效果如下。
上面了解了短信的处理大致的内容，我们就需要整合它进行短信的发送了。
首先我们为了方便，需要在项目的配置文件.env 中增加配置文件，方便统一使用。
为了能够在多个地方使用，我们对短信发送的处理进行简单的封装一下，如下所示的辅助类，初始化的时候，提供相应的配置的参数即可。
from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_dysmsapi20170525 import models as dysmsapi_20170525_models from alibabacloud_tea_util import models as util_models from alibabacloud_tea_util.client import Client as UtilClient from typing import List, Dict, Any class SMSHelper: """ 阿里云短信服务工具类 """ def __init__(self, access_key_id: str, access_key_secret: str, sign_name: str, template_code: str, endpoint : str = 'dysmsapi.aliyuncs.com' ): """ 初始化阿里云短信服务客户端 :param access_key_id: 阿里云访问密钥 ID :param access_key_secret: 阿里云访问密钥 Secret :param endpoint: 阿里云 API 网关地址 :param sign_name: 短信签名 :param template_code: 短信模板代码 """ if not access_key_id or not access_key_secret: raise ValueError('没有设置阿里云访问凭据，请设置环境变量或在代码中设置') config = open_api_models.Config( access_key_id=access_key_id, access_key_secret=access_key_secret, endpoint=endpoint) self.client = Dysmsapi20170525Client(config) self.sign_name = sign_name self.template_code = template_code
初始化后，就可以调用参数进行发送短信了
def send_sms(self, phone_numbers: str, template_param: Dict[str, Any] = None) -> Dict[str, Any]: """ 使用阿里云 SMS 服务发送短信 Args: phone_numbers: 短信接收号码 template_param: 短信模板参数，字典形式 Returns: 短信发送结果，字典形式 """ # 创建 request 对象 send_sms_request = dysmsapi_20170525_models.SendSmsRequest( phone_numbers=phone_numbers, sign_name= self.sign_name, template_code=self.template_code, template_param=str(template_param) if template_param else None ) response = {} try: # 发送短信 result = self.client.send_sms_with_options(send_sms_request, util_models.RuntimeOptions()) # 转换结果为字典形式 response = { 'success': True, 'message': 'SMS 发送成功', 'request_id': result.body.request_id, 'code': result.body.code, 'message': result.body.message, 'biz_id': result.body.biz_id } except Exception as error: response = { 'success': False, 'message': str(error.message) if hasattr(error, 'message') else str(error), 'recommend': error.data.get("Recommend") if hasattr(error, 'data') else None } return response
完成了上面的简单封装，就可以再API的控制器端进行短信处理了。
如我们在登录login的EndPoint（类似C#的控制器类）中需要先初始化短信的服务辅助类。
上面也介绍过，短信主要就是解决BS端 或者H5端的身份登录及密码修改、重置等重要处理的通知。
我们在以手机号码和短信号码登录的时候，需要先发送短信，然后短信会在服务端通过Redis的缓存驻留几分钟，这几分钟内容，通过手机和验证码即可登录，登录后验证码失效，如果超时验证码也失效。
通过手机号码发送短信的过程如下所示。
@router.post( "/send-login-smscode", summary="发送登录验证码", response_model=AjaxResponse[CommonResult | None], ) async def send_login_smscode( input: Annotated[PhoneCaptchaModel, Body()], request: Request, db: AsyncSession = Depends(get_db), ): ip = await get_ip(request) # 校验手机号码是否合法 if not input.phonenumber.isdigit() or len(input.phonenumber) != 11: raise CustomExceptionBase(detail="手机号码格式不正确") # 校验用户是否存在 user = await user_crud.get_by_column(db, "mobilephone", input.phonenumber) if not user: raise CustomExceptionBase(detail="用户不存在") # 生成6位数字验证码 code = RandomUtil.random_digit_string(6) # 发送短信验证码 res = sms_helper.send_sms( phone_numbers=input.phonenumber, template_param={"code": code}, ) success = res.get("success", False) == True message = res.get("message", "") if success: #以手机号码作为键存储验证码缓存 cache_key = input.phonenumber.strip() cache_item = SmsLoginCodeCacheItem( phonenumber=input.phonenumber.strip(), code=code, ).model_dump() redis_helper = RedisHelper(client = redis_client) await redis_helper.set(cache_key, cache_item, 60 * settings.SMS_EXPIRED_MINUTES) # 默认5分钟过期 # 短信验证码发送结果 result = CommonResult(success=success, errormessage=message) return AjaxResponse(result)
上面短信发送后，号码机验证码会驻留在Redis的缓存中一段时间，那么此时如果使用手机验证码进行登录即可匹配到。
@router.post( "/authenticate-byphone", summary="手机短信验证码登录授权处理", response_model=AjaxResponse[AuthenticateResultDto], ) async def authenticate_by_phone( input: Annotated[PhoneCaptchaModel, Body()], request: Request, db: AsyncSession = Depends(get_db), ): ip = await get_ip(request) auth_result = AuthenticateResultDto(ip=ip, success=False) # 从缓存中获取验证码 redis_helper = RedisHelper(client=redis_client) cache_key = input.phonenumber.strip() if not cache_key.isdigit() or len(cache_key) != 11: raise CustomExceptionBase(detail="手机号码格式不正确") # 校验验证码是否正确 if not input.smscode.isdigit() or len(input.smscode) != 6: raise CustomExceptionBase(detail="验证码格式不正确") cache_item = await redis_helper.get(cache_key) if not cache_item: raise CustomExceptionBase(detail="验证码已过期或不存在，请重新获取") cache_item = SmsLoginCodeCacheItem(**cache_item) if cache_item.code != input.smscode: raise CustomExceptionBase(detail="验证码不正确，请重新输入") # 校验用户是否存在 user = await user_crud.get_by_column(db, "mobilephone", input.phonenumber) if not user: raise CustomExceptionBase(detail="用户不存在") # 验证码正确，继续登录流程 if cache_item and user: #获取用户角色类型 auth_result.roletype = await role_crud.get_user_roletype( db, user.id ) # 超级管理员、系统管理员、其他 # 根据用户身份生成tokenresult auth_result.expires = int((datetime.utcnow() + timedelta(seconds=settings.TOKEN_EXPIRE_SECONDS)).timestamp()) auth_result.userid = user.id auth_result.name = user.name auth_result.success = True auth_result.accesstoken = await generate_token(vars(user), role_type=auth_result.roletype) #移除缓存短信键值 await redis_helper.delete(cache_key) else: auth_result.error = "登录失败，无法生成令牌" return AjaxResponse(auth_result)
其他的重置密码，修改密码，修改重要信息等通知的处理也是类似的处理过程，不在赘述。
我们通过pip命令安装fastapi_mail组件进行邮件发送的处理。
邮件发送，一般也是基于模板文件的方式，通过对模板文件的变量进行变化，实现内容的发送过程。
我们的模板路径如下所示，里面有类似下面的几个模板文件，如果需要更多场合的右键，可以进行不同的模板编写即可。
# templates/ # email/ # welcome.html # reset_password.html
增加一个EmailHelper.py的辅助类简单处理下，方便后续的邮件发送处理。
from typing import List, Optional, Dict from fastapi.templating import Jinja2Templates from fastapi_mail import FastMail, MessageSchema, ConnectionConfig, MessageType from fastapi import BackgroundTasks from jinja2 import Environment, FileSystemLoader, select_autoescape class EmailHelper: """ 邮件发送辅助类，基于 fastapi-mail """ def __init__( self, username: str, password: str, mail_from: str, server: str = "smtp.example.com", port: int = 587, use_tls: bool = True, use_ssl: bool = False, template_folder: str = "app/templates/email", enable_template_cache: bool = True, # 开关：是否启用模板缓存 ): self.conf = ConnectionConfig( MAIL_USERNAME=username, MAIL_PASSWORD=password, MAIL_FROM=mail_from, MAIL_PORT=port, MAIL_SERVER=server, MAIL_SSL_TLS=use_ssl, MAIL_STARTTLS=True if use_tls else False, USE_CREDENTIALS=True, ) self.fm = FastMail(self.conf) # 模板目录 # self.templates = Jinja2Templates(directory=template_folder) # 增加 模板缓存 / 热更新开关，这样在 开发环境下可以实时修改模板生效，在 生产环境下则启用缓存提升性能。 # Jinja2 环境 self.env = Environment( loader=FileSystemLoader(template_folder), autoescape=select_autoescape(["html", "xml"]), cache_size=50 if enable_template_cache else 0, # 0 = 每次重新加载模板 ) async def send_email( self, subject: str, recipients: List[str], body: str, subtype: MessageType = MessageType.plain, background_tasks: Optional[BackgroundTasks] = None, attachments: Optional[List[str]] = None, ): """ 发送邮件（支持同步调用和后台任务） :param subject: 邮件主题 :param recipients: 收件人列表 :param body: 邮件正文 :param subtype: 内容类型（plain 或 html） :param background_tasks: FastAPI BackgroundTasks（可选） :param attachments: 附件路径列表（可选） """ message = MessageSchema( subject=subject, recipients=recipients, body=body, subtype=subtype, attachments=attachments or [] # 如果是 None，自动变成 [] ) if background_tasks: background_tasks.add_task(self.fm.send_message, message) else: await self.fm.send_message(message) async def send_template_email( self, subject: str, recipients: List[str], template_name: str, context: Dict, background_tasks: Optional[BackgroundTasks] = None, attachments: Optional[List[str]] = None, ): """ 发送基于 Jinja2 模板的邮件 :param subject: 邮件主题 :param recipients: 收件人列表 :param template_name: 模板文件名 (如 welcome.html) :param context: 模板变量 """ # 渲染模板 # template = self.templates.get_template(template_name) template = self.env.get_template(template_name) body = template.render(**context) await self.send_email( subject=subject, recipients=recipients, body=body, subtype=MessageType.html, background_tasks=background_tasks, attachments=attachments or [] # 如果是 None，自动变成 [] )
我们在一个独立的EndPoint的类中提供邮件发送的处理API。
初始化接口如下所示
两个利用邮件模板发送邮件的例子如下所示。
以上就是基于FastAPI的Python开发框架后端，增加阿里云短信和邮件发送通知处理的相关实现过程，希望对你有所帮助。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。