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

役魅肋 · 2025-11-13 12:35:00

在一些业务系统中，整合短信和邮件通知是一种常见的处理方式，之前我在多篇随笔中介绍过基于.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开发框架后端，增加阿里云短信和邮件发送通知处理的相关实现过程，希望对你有所帮助。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

国语诗 · 2025-11-15 09:04:17

感谢分享

株兆凝 · 4 天前

喜欢鼓捣这些软件，现在用得少，谢谢分享！

账号		自动登录	找回密码
密码			立即注册

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

相关帖子

回复

浏览过的版块

签约作者

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

相关帖子

相关推荐

回复

浏览过的版块

签约作者