在基于FastAPI的Python开发框架后端，增加阿里云短信和邮件发送通知处理

在一些业务系统中，整合短信和邮件通知是一种常见的处理方式，之前我在多篇随笔中介绍过基于.NET的整合处理，本篇随笔介绍基于FastAPI的Python开发框架后端，增加阿里云短信和邮件发送通知处理。

1、短信通知处理的介绍

之前我在多篇随笔中介绍过基于.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"
 }
}

短信真实接收到的效果如下。

2、在基于FastAPI的Python开发框架后端整合接口发送短信

上面了解了短信的处理大致的内容，我们就需要整合它进行短信的发送了。

首先我们为了方便，需要在项目的配置文件.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)

其他的重置密码，修改密码，修改重要信息等通知的处理也是类似的处理过程，不在赘述。

3、在基于FastAPI的Python开发框架中整合邮件发送

我们通过pip命令安装fastapi_mail组件进行邮件发送的处理。

pip install 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开发框架后端，增加阿里云短信和邮件发送通知处理的相关实现过程，希望对你有所帮助。