博客 / 詳情

返回

在基於FastAPI的Python開發框架後端，增加阿里雲短信和郵件發送通知處理

在一些業務系統中，整合短信和郵件通知是一種常見的處理方式，之前我在多篇隨筆中介紹過基於.NET的整合處理，本篇隨筆介紹基於FastAPI的Python開發框架後端，增加阿里雲短信和郵件發送通知處理。

1、短信通知處理的介紹

之前我在多篇隨筆中介紹過基於.NET的《SqlSugar開發框架》中整合過短信接入的內容：《使用阿里雲的短信服務發送短信》、《基於SqlSugar的開發框架循序漸進介紹（17）-- 基於CSRedis實現緩存的處理》、《循序漸進VUE+Element 前端應用開發(32）--- 手機短信動態碼登陸處理》。

短信通知，一般用於系統的登錄，或者對重要數據變更的身份確認，各個平台都相關的短信接口，不過好像華為服務器已經不提供短信接入了，阿里雲還可以，或者也可以找一些其他的短信服務商，基本上都會提供相應的接口或者SDK，對接還是很方便的。

本篇隨筆基於阿里雲的短信接口進行對接短信通知，也主要就是解決BS端 或者H5端的身份登錄及密碼修改、重置等重要處理的通知。

對於Python開發來説，基於阿里雲的短信處理，可以使用它的 alibabacloud_dysmsapi20170525 SDK包來進行對接，雖然這個包時間上比較老，好像也沒有看到更新的SDK了。

image

使用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 中增加配置文件，方便統一使用。

image

 為了能夠在多個地方使用，我們對短信發送的處理進行簡單的封裝一下，如下所示的輔助類，初始化的時候，提供相應的配置的參數即可。

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#的控制器類）中需要先初始化短信的服務輔助類。

image

上面也介紹過，短信主要就是解決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。

初始化接口如下所示

image

兩個利用郵件模板發送郵件的例子如下所示。

image

 

以上就是基於FastAPI的Python開發框架後端，增加阿里雲短信和郵件發送通知處理的相關實現過程，希望對你有所幫助。

後端
user avatar u_16213589 頭像 azonips314 頭像 u_16213461 頭像 prepared 頭像 13917911249 頭像 pudongping 頭像 tingtr 頭像 mylxsw 頭像 yutou_5c10e66caa840 頭像 ziwupython 頭像 fanjiapeng 頭像 zbooksea 頭像
37 位用戶收藏了這個故事！

發佈 評論

Some HTML is okay.