主页
文章
分类
标签
关于
FastAPI 邮箱发送功能
发布于: 2025-12-6   更新于: 2025-12-6   收录于: Fastapi
文章字数: 2045   阅读时间: 5 分钟   阅读量:

前置介绍

本实验参考知了传课的最新教程实现，利用fastapi-mail发送邮箱验证码，实验采用qq邮箱来进行邮箱的发送和接受

安装插件

1

pip install fastapi-mail

邮箱配置

  • 首先打开QQ邮箱的首页 点击左上角 设置
示例
  • 然后点开左边的账号与安全
示例
  • 再进入安全设置
示例

  • 我们绑定完手机之后，去生成授权码 示例

  • 系统将生成一个 16 位授权码 示例

注意：授权码用于 SMTP 身份验证，不是邮箱登录密码，每次生成的都不一样，如果被别人弄走了，重新生成一个即可。

Fastapi-Mail配置

配置模型

在自己的配置文件（我这里是settings），放置自己的配置，推荐使用pydantic_setting的BaseSettings：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# settings.py
from pydantic_settings import BaseSettings

class MailSettings(BaseSettings):
    MAIL_USERNAME: str
    MAIL_PASSWORD: str
    MAIL_FROM: str
    MAIL_PORT: int = 587
    MAIL_SERVER: str = "smtp.qq.com"
    MAIL_FROM_NAME: str = "FastAPI Mail Service"
    MAIL_STARTTLS: bool = True
    MAIL_SSL_TLS: bool = False
    USE_CREDENTIALS: bool = True
    VALIDATE_CERTS: bool = True

    class Config:
        env_file = ".env"

创建环境变量文件

同时创建.env文件：

1
2
3

MAIL_USERNAME=your_email@qq.com
MAIL_PASSWORD=your_16_digit_auth_code
MAIL_FROM=your_email@qq.com

创建 FastMail 实例与依赖注入

创建工厂函数

先写一个FastMail实例的工厂函数，每次调用都返回一个新的实例，防止阻塞

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from fastapi_mail import FastMail, ConnectionConfig
from pydantic import SecretStr, EmailStr
from settings import base_settings

def create_mail_instance() -> FastMail:

    mail_config = ConnectionConfig(
        MAIL_USERNAME=base_settings.MAIL_USERNAME,
        MAIL_PASSWORD=base_settings.MAIL_PASSWORD,
        MAIL_FROM=base_settings.MAIL_FROM,
        MAIL_PORT=base_settings.MAIL_PORT,
        MAIL_SERVER=base_settings.MAIL_SERVER,
        MAIL_FROM_NAME=base_settings.MAIL_FROM_NAME,
        MAIL_STARTTLS=base_settings.MAIL_STARTTLS,
        MAIL_SSL_TLS=base_settings.MAIL_SSL_TLS,
        USE_CREDENTIALS=True,
        VALIDATE_CERTS=True,
    )

    return FastMail(mail_config)

创建依赖函数

根据工厂函数创建依赖

1
2
3
4
5
6
7

from fastapi_mail import FastMail

from models import AsyncSessionFactory
from core.mail import create_mail_instance

async def get_mail() -> FastMail:
    return create_mail_instance()

测试接口

写一个api测试一下邮件发送：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from fastapi import FastAPI, Depends
from fastapi_mail import FastMail, MessageSchema, MessageType
from aiosmtplib import SMTPResponseException

from dependencies import get_mail

app = FastAPI()
@app.get("/mail/test")
async def test_mail(email: str, mail: FastMail = Depends(get_mail)):
    message = MessageSchema(
        subject="测试邮件",
        recipients=[email],
        body=f"HELLO {email}",
        subtype=MessageType.plain
    )

    await mail.send_message(message)

    return {"msg": "发送成功"}

这里引入一个自带的pydatic模型MessageSchema用于数据校验，具体就是这四个东西：

  • subject=“测试邮件”： 邮件主题
  • recipients=[email]： 收件人列表
  • body=f"HELLO {email}"： 邮件正文
  • subtype=MessageType.plain：内容类型（plain 或 html）

注意 QQ 邮箱 SMTP 限制:

  • 使用端口 587（STARTTLS）或 465（SSL）。本文配置使用 587。
  • 每日发送限额约为 500 封，超出后会被临时限制。

fastapi-mail 已知问题

在部分 SMTP 服务器（如 QQ 邮箱）关闭连接时，会返回非标准字节流，导致底层 aiosmtplib 抛出 code=-1 的异常。此为协议兼容性问题，邮件已成功投递，可安全忽略该特定异常。 alt text

优化错误处理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

@app.get("/mail/test")
async def test_mail(email: str, mail: FastMail = Depends(get_mail)):
    message = MessageSchema(
        subject="测试邮件",
        recipients=[email],
        body=f"HELLO {email}",
        subtype=MessageType.plain
    )
    try:
        await mail.send_message(message)
    except SMTPResponseException as e:
        if e.code == -1 and b"\\x00\\x00\\x00" in str(e).encode():
            print("忽略QQ邮箱SMTP关闭阶段的非标准式响应，邮箱已经发送成功")
    
    return {"msg": "发送成功"}

在获取到对应错误的时候，忽略一下即可

验证码生成

数据模型

先创建一个Emailcode数据库模型：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

class Base(DeclarativeBase):
    """基类：通用字段"""
    create_time: Mapped[datetime] = mapped_column(
        DateTime, 
        default=func.now(), 
        comment="创建时间"
    )
    update_time: Mapped[datetime] = mapped_column(
        DateTime, 
        default=func.now(), 
        onupdate=func.now(), 
        comment="更新时间"
    )

class EmailCode(Base):
    __tablename__ = "email_code"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(100))
    code: Mapped[str] = mapped_column(String(10))

数据访问层

这里由于使用数据库存储生成的验证码，所以额外开一个视图来编写emailcode相关的数据库操作：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# user_repo.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional
from datetime import datetime, timedelta

from models.user import EmailCode

class EmailCodeRepository:

    def __init__(self, session: AsyncSession):
        self.session = session # 传入数据库会话

    async def create(self, email: str, code: str):
        """
        在数据库中存储code，用于后续验证
        """
        email_code = EmailCode(email=email, code=code)
        self.session.add(email_code)
        return email_code
        
    async def check_mail_code(self, email: str, code: str) -> bool:
        """
        在数据库中检查code是否存在以及是否过期
        """
        stmt = select(EmailCode).where(EmailCode.email == email, EmailCode.code == code)
        email_code: Optional[EmailCode] = await self.session.scalar(stmt)
        print(email_code)
        if email_code is None:
            print("验证码不存在")
            return False
        if (datetime.now() - email_code.create_time) > timedelta(minutes=10): # 设置验证码有效期
            print("验证码已过期")
            return False
        return True

验证码发送路由

然后编写code路由，用于向指定邮箱发送验证码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

class ResponseOut(BaseModel):
    """
    用于一些视图函数只要返回操作结果的模型
    """
    result: Annotated[Literal["success", "failure"], Field("success", description="操作的结果")] 

@router.get("/code", response_model=ResponseOut)
async def get_mail_code(
    email: Annotated[EmailStr, Query(description="邮箱")], 
    mail: Annotated[FastMail, Depends(get_mail)],
    session: Annotated[AsyncSession, Depends(get_session)]
):
    """
    生成并发送四位验证码
    """
    # 生成4位数字验证码
    source = string.digits * 4 # 4*(1~10) 
    code = "".join(random.sample(source, 4)) # 从40个数字中随机采样四次
    # 创建消息对象
    message = MessageSchema(
        subject="AiName注册验证码",
        recipients=[email],
        body=f"您的验证码是：{code}",
        subtype=MessageType.plain
    )

    # 创建操作视图
    email_code_repo = EmailCodeRepository(session = session)
    # fastmail发送邮件
    try:
        await mail.send_message(message)
        await email_code_repo.create(str(email), code)
        await session.commit() # 这里我手动提交事务 与原视频不一样 防止事务重复开启
    except SMTPResponseException as e:
        if e.code == -1 and b"\\x00\\x00\\x00" in str(e).encode():
            # 忽略 QQ 邮箱关闭连接时的非标准响应
            print("忽略QQ邮箱SMTP关闭阶段的非标准式响应，邮箱已经发送成功")
            await email_code_repo.create(str(email), code)  
            await session.commit()
        else:
            await session.rollback() 
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="发送邮件失败"
            )
        
    return ResponseOut(result="success")

这样一个邮箱验证码发送的接口就写好了，可以在docs里尝试一下

Python
Fastapi
uni-app的简单实验