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

布衣人家 發表於 2025-11-13 11:53:00

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

在一些业务系统中，整合短信和邮件通知是一种常见的处理方式，之前我在多篇随笔中介绍过基于.NET的整合处理，本篇随笔介绍基于FastAPI的Python开发框架后端，增加阿里云短信和邮件发送通知处理。
<h3>1、短信通知处理的介绍</h3>
之前我在多篇随笔中介绍过基于.NET的《SqlSugar开发框架》中整合过短信接入的内容：《使用阿里云的短信服务发送短信》、《基于SqlSugar的开发框架循序渐进介绍（17）-- 基于CSRedis实现缓存的处理》、《循序渐进VUE+Element 前端应用开发(32）--- 手机短信动态码登陆处理》。
短信通知，一般用于系统的登录，或者对重要数据变更的身份确认，各个平台都相关的短信接口，不过好像华为服务器已经不提供短信接入了，阿里云还可以，或者也可以找一些其他的短信服务商，基本上都会提供相应的接口或者SDK，对接还是很方便的。
本篇随笔基于阿里云的短信接口进行对接短信通知，也主要就是解决BS端或者H5端的身份登录及密码修改、重置等重要处理的通知。
对于Python开发来说，基于阿里云的短信处理，可以使用它的 alibabacloud_dysmsapi20170525 SDK包来进行对接，虽然这个包时间上比较老，好像也没有看到更新的SDK了。
<img src="https://img2024.cnblogs.com/blog/8867/202511/8867-20251113111817633-1978375735.png" alt="image" loading="lazy">
使用pip 命令安装了该SDK即可。
<div class="cnblogs_code">
<pre>pip install alibabacloud_dysmsapi20170525</pre>
</div>
我们使用阿里云API发送短信，一般需要提供下面的身份信息。
使用阿里云的短信服务，需要注册登录自己的阿里云控制台，然后进入AccessKeys的处理界面
<img src="https://img2018.cnblogs.com/blog/8867/201904/8867-20190410105810429-492056133.png" alt="">
这里我们获取到AccessKey ID 和Access Key Secret两个关键信息，需要用在数据签名的里面的。
发送接口还需要提供下面的的一些信息，包括必要的手机号码，签名，服务器端模板代码，以及短信码等必要信息。
签名，一般为我们短信提示的公司名称，如【广州爱奇迪】这样的字样。
<img src="https://img2018.cnblogs.com/blog/8867/201904/8867-20190410110210242-1598268385.png" alt="" class="medium-zoom-image">
服务器端模板代码，阿里云默认提供了一些基础模板，我们可以从中选择。
<img src="https://img2018.cnblogs.com/blog/8867/201904/8867-20190410112024695-1167915169.png" alt="" class="medium-zoom-image">
如下具体接口请求需要提供的JSON数据。
<div class="cnblogs_code">
<pre>{
"phone_numbers": "13800138000",
"sign_name": "YourSignName",
"template_code": "SMS_123456789",
"template_param": {
"code": "123456"
}
}</pre>
</div>
短信真实接收到的效果如下。
<img src="https://img2018.cnblogs.com/blog/8867/201904/8867-20190410113014515-1725476367.png" alt="" class="medium-zoom-image">
 
<h3>2、在基于FastAPI的Python开发框架后端整合接口发送短信</h3>
上面了解了短信的处理大致的内容，我们就需要整合它进行短信的发送了。
首先我们为了方便，需要在项目的配置文件.env 中增加配置文件，方便统一使用。
<img src="https://img2024.cnblogs.com/blog/8867/202511/8867-20251113113046388-1382017109.png" alt="image" loading="lazy">
 为了能够在多个地方使用，我们对短信发送的处理进行简单的封装一下，如下所示的辅助类，初始化的时候，提供相应的配置的参数即可。
<div class="cnblogs_code">
<pre>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</pre>
</div>
初始化后，就可以调用参数进行发送短信了
<div class="cnblogs_code">
<pre> def send_sms(self, phone_numbers: str, template_param: Dict = None) -> Dict:
 """
 使用阿里云 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</pre>
</div>
完成了上面的简单封装，就可以再API的控制器端进行短信处理了。
如我们在登录login的EndPoint（类似C#的控制器类）中需要先初始化短信的服务辅助类。
<img src="https://img2024.cnblogs.com/blog/8867/202511/8867-20251113113642840-161364699.png" alt="image" loading="lazy">
上面也介绍过，短信主要就是解决BS端或者H5端的身份登录及密码修改、重置等重要处理的通知。
我们在以手机号码和短信号码登录的时候，需要先发送短信，然后短信会在服务端通过Redis的缓存驻留几分钟，这几分钟内容，通过手机和验证码即可登录，登录后验证码失效，如果超时验证码也失效。
通过手机号码发送短信的过程如下所示。
<div class="cnblogs_code">
<pre>@router.post(
"/send-login-smscode",
summary="发送登录验证码",
response_model=AjaxResponse,
)
async def send_login_smscode(
input: Annotated,
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) </pre>
</div>
上面短信发送后，号码机验证码会驻留在Redis的缓存中一段时间，那么此时如果使用手机验证码进行登录即可匹配到。
<div class="cnblogs_code">
<pre>@router.post(
"/authenticate-byphone",
summary="手机短信验证码登录授权处理",
response_model=AjaxResponse,
)
async def authenticate_by_phone(
input: Annotated,
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)</pre>
</div>
其他的重置密码，修改密码，修改重要信息等通知的处理也是类似的处理过程，不在赘述。
 
<h3>3、在基于FastAPI的Python开发框架中整合邮件发送</h3>
我们通过pip命令安装fastapi_mail组件进行邮件发送的处理。
<div class="cnblogs_code">
<pre>pip installfastapi_mail</pre>
</div>
邮件发送，一般也是基于模板文件的方式，通过对模板文件的变量进行变化，实现内容的发送过程。
我们的模板路径如下所示，里面有类似下面的几个模板文件，如果需要更多场合的右键，可以进行不同的模板编写即可。
<div class="cnblogs_code">
<pre># templates/
# email/
# welcome.html
# reset_password.html</pre>
</div>
增加一个EmailHelper.py的辅助类简单处理下，方便后续的邮件发送处理。
<div class="cnblogs_code">
<pre>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,
 body: str,
 subtype: MessageType = MessageType.plain,
 background_tasks: Optional = None,
 attachments: Optional] = 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,
 template_name: str,
 context: Dict,
 background_tasks: Optional = None,
 attachments: Optional] = 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，自动变成 []
 )</pre>
</div>
我们在一个独立的EndPoint的类中提供邮件发送的处理API。
初始化接口如下所示
<img src="https://img2024.cnblogs.com/blog/8867/202511/8867-20251113114948608-186272654.png" alt="image" loading="lazy">
两个利用邮件模板发送邮件的例子如下所示。
<img src="https://img2024.cnblogs.com/blog/8867/202511/8867-20251113115236925-1467434832.png" alt="image" loading="lazy">
 
以上就是基于FastAPI的Python开发框架后端，增加阿里云短信和邮件发送通知处理的相关实现过程，希望对你有所帮助。

</div>
<div id="MySignature" role="contentinfo">
<div style="border-right-color: #cccccc; border-right-width: 1px; border-right-style: solid; padding-right: 5px; border-top-color: #cccccc; border-top-width: 1px; border-top-style: solid; padding-left: 4px; font-size: 13px; padding-bottom: 4px; border-left-color: #cccccc; border-left-width: 1px; border-left-style: solid; width: 98%; padding-top: 4px; border-bottom-color: #cccccc; border-bottom-width: 1px; border-bottom-style: solid; background-color: #eeeeee;">
<img src="http://www.cnblogs.com/Images/OutliningIndicators/None.gif" align="top" alt>

专注于代码生成工具、.Net/Python 框架架构及软件开发，以及各种Vue.js的前端技术应用。著有Winform开发框架/混合式开发框架、微信开发框架、Bootstrap开发框架、ABP开发框架、SqlSugar开发框架、Python开发框架等框架产品。
 转载请注明出处：撰写人：伍华聪 http://www.iqidi.com </div> 
来源：https://www.cnblogs.com/wuhuacong/p/19217500

頁: [1]

琼殿技术论坛's Archiver

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