Перейти к контенту

FastAPI 7. Электронная почта, подтверждение регистрации, Celery и Redis

Сервис на FastAPI • Иван Ашихмин • • 1704
Python Гайды email itsdangerous FastAPI подтверждение регистрации отправка почты

В этой статье мы напишем подтверждение регистрации по электронной почте. Для этого подключим Celery и настроим фоновую отправку почты.

FastAPI 7. Электронная почта, подтверждение регистрации, Celery и Redis
Сервис на FastAPI • Иван Ашихмин • • 1704

Зарегистрированный пользователь хорошо, а подтверждённый по почте ещё лучше. Регистрироваться на сайте могут боты в больших количествах, что может негативно сказаться на различных аспектах работы проекта или банально портить статистику, не отражая истинной картины количества пользователей.

Важно: эта статья — часть серии «Веб-сервис на FastAPI». Она опирается на код и архитектурные решения, описанные в предыдущих материалах. Если вы попали сюда напрямую, некоторые места могут показаться неполными или слишком короткими — это нормально, просто в рамках цикла мы практически не повторяем уже разобранное.

 

Нам понадобится

Для всего этого нам понадобятся следующие инструменты:

  • Celery и Redis - Для выполнения задачи отправки почты в фоновом режиме.
  • smtplib - Библиотека для работы с SMTP-сервером. Именно она будет подключаться к почтовому серверу и отправлять почту. Входит в стандартную библиотеку Python.
  • jinja2 - Библиотека для рендеринга HTML-шаблона. Необходима для отправки письма с HTML-форматированием. Входит в состав FastAPI.
  • itsdangerous - Библиотека для безопасной работы с данными, которые доступны в открытом виде, например, токен для подтверждения регистрации.

Уставим необходимые библиотеки, выполнив команду:

poetry add celery redis itsdangerous

Подробно прочитать про Celery и Redis можно в статье "Django 43. Подключаем Celery и Redis для фоновой отправки почты", в этом не буду углубляться в подробности, дабы не растягивать статью.

 

Новые переменные окружения.

Откроем .env-файл и добавим новые переменные окружения:

  • EMAIL_HOST - указываем хост почтового сервера.
  • EMAIL_PORT - указываем порт почтового сервера.
  • EMAIL_USERNAME - указываем имя пользователя почтового сервера, обычно это адрес электронной почты.
  • EMAIL_PASSWORD - указываем пароль почтового сервера.
  • REDIS_HOST - указываем адрес хоста Redis.
  • REDIS_PORT - указываем порт Redis.
  • REDIS_DB - указываем номер используемой базы данных в Redis. Доступно от 0 до 15-ти.
  • FRONTEND_URL - в этом поле будущем пропишем URL фронтенда, но пока его нет, а подтверждать почту надо, пропишем URL бэкэнда, например, http://127.0.0.1:8000/api/v1.

Пример:

EMAIL_HOST=smtp.yandex.ru
EMAIL_PORT=465
EMAIL_USERNAME=info@yandex.ru
EMAIL_PASSWORD=12345
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_DB=0
FRONTEND_URL=http://127.0.0.1:8000/api/v1

Теперь откроем файл settings.py в пакете core.

По аналогии с классом DBSettings, создадим два класса: EmailSettings и RedisSettings.

Класс EmailSettings

Тут всё просто, добавляем поля из переменных окружения связанные с почтой, а также поле для чтения .env-файла:

class EmailSettings(BaseSettings):  
    email_host: str  
    email_port: int  
    email_username: str  
    email_password: SecretStr  

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf8", extra="ignore")

Класс RedisSettings

Точно также прописываем три поля из .env, поле для чтения переменных окружения. Затем ниже прописываем property-функцию, формирующую Redis-URI:

class RedisSettings(BaseSettings):  
    redis_host: str  
    redis_port: int  
    redis_db: int  

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf8", extra="ignore")  

    @property  
    def redis_url(self):  
        return f"redis://{self.redis_host}:{self.redis_port}/{self.redis_db}"

Класс Settings

Теперь в него необходимо добавить созданные ранее классы, а также новые поля:

  • email_settings - объект класса EmailSettings
  • redis_settings - объект класса RedisSettings
  • templates_dir - название директории с HTML-шаблонами, обычно это templates.
  • frontend_url - строковое поле.

Код изменений:

from pydantic import SecretStr  
from pydantic_settings import BaseSettings, SettingsConfigDict


class EmailSettings(BaseSettings):  
    email_host: str  
    email_port: int  
    email_username: str  
    email_password: SecretStr  

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf8", extra="ignore")  


class RedisSettings(BaseSettings):  
    redis_host: str  
    redis_port: int  
    redis_db: int  

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf8", extra="ignore")  

    @property  
    def redis_url(self):  
        return f"redis://{self.redis_host}:{self.redis_port}/{self.redis_db}"  


class Settings(BaseSettings):  
    db_settings: DBSettings = DBSettings()  
    email_settings: EmailSettings = EmailSettings()  
    redis_settings: RedisSettings = RedisSettings()  
    secret_key: SecretStr  
    templates_dir: str = "templates"  
    frontend_url: str  

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf8", extra="ignore")  


settings = Settings()

 

Celery-приложение.

После подготовки, необходимо определить Celery-приложение.

В пакете Core создадим файл celery_config.py со следующим содержимым:

from celery import Celery  

from lkeep.core.settings import settings  

celery_app = Celery(main="lkeep", broker=settings.redis_settings.redis_url, backend=settings.redis_settings.redis_url)  

celery_app.autodiscover_tasks(packages=["lkeep.apps"])

Там же в пакете core откроем файл __init__.py и добавим импорт и экспорт Celery-приложения. Это необходимо для корректной работы Celery.

from .celery_config import celery_app  

__all__ = ['celery_app']

Разберём код:

Инициализируем переменную celery_app в которой определяем экземпляр класса Celery. В конструктор передаём следующие аргументы:

  • main - Название Celery-приложения.
  • broker - Брокер сообщений, передаём сюда redis_url.
  • backend - БД для записи результатов выполнения задач, тоже указываем redis_url или другую БД.

Далее у переменной celery_app, вызываем метод .autodiscover_tasks, который автоматически будет искать задачи для выполнения. Внутрь передаём аргумент packages, в котором указываем список строк. В строке прописываем имена пакетов в которых искать задачи, в нашем случае это lkeep.apps. Celery будет искать файлы tasks.py в пакете apps пакета lkeep.

 

Celery-задача отправки почты.

Нам нужна функция, в которой будет определена отправка почты. В пакете auth создадим файл tasks.py.

В этом файле создадим функцию send_confirmation_email, принимающую два строковых аргумента: to_email и token. И обернём функцию в декоратор @shared_task для того, чтобы Celery воспринимал её как задачу.

Отправка тесктового письма.

Сперва рассмотрим отправку простого текстового письма.

Прописываем переменную confirmation_url. В ней при помощи f-строки формируем адрес для подтверждения регистрации. Состоящий из:

  • Адреса сайта - Поле frontend_url в классе Settings, в нашем случае там адрес локального сервера.
  • Маршрута подтверждения регистрации.
  • Query-параметра token с уникальной строкой.

Получится вот такая строка:

confirmation_url = f"{settings.frontend_url}/auth/register_confirm?token={token}"

Также прописываем переменную text с многострочной строкой текста письма и ссылкой подтверждения почты из созданной ранее переменной, например:

text = f"""Спасибо за регистрацию!  
Для подтверждения регистрации перейдите по ссылке: {confirmation_url}  
"""

Далее создаём объект письма и присваиваем ему параметры:

  1. Создаём переменную message и объявляем её экземпляром класса EmailMessage.
  2. У переменной message вызываем метод .set_content(), передавая в него текст письма из переменной text.
  3. В переменную message по ключу From, присваиваем адрес электронной отправителя.
  4. В переменную message по ключу To, присваиваем адрес электронной получателя.
  5. В переменную message по ключу Subject, присваиваем письму тему.

Затем при помощи контекстного менеджера with вызываем класс smtplib.SMTP_SSL, в который передаём два аргумента:

  • host - Хост для подключения к почтовому серверу.
  • port - Порт для подключения к почтовому серверу.

Присваиваем открытому классу имя smtp.

Внутри контекстного менеджера у переменной smtp вызываем метод .login в который передаём два аргумента:

  • user - Имя пользователя учётной записи в почтовом сервере, обычно это адрес электронной почты.
  • password - Пароль от учётной записи в почтовом сервере.

Наконец, вызываем у переменной smtp метод send_message, передав в него аргумент msg.

Код файла:

import smtplib
from email.message import EmailMessage

from celery import shared_task

from lkeep.core.settings import settings


@shared_task
def send_confirmation_email(to_email: str, token: str) -> None:
    confirmation_url = f"{settings.frontend_url}/auth/register_confirm?token={token}"

    text = f"""Спасибо за регистрацию!
Для подтверждения регистрации перейдите по ссылке: {confirmation_url}
"""

    message = EmailMessage()
    message.set_content(text)
    message["From"] = settings.email_settings.email_username
    message["To"] = to_email
    message["Subject"] = "Подтверждение регистрации"

    with smtplib.SMTP_SSL(host=settings.email_settings.email_host, port=settings.email_settings.email_port) as smtp:
        smtp.login(
            user=settings.email_settings.email_username,
            password=settings.email_settings.email_password.get_secret_value(),
        )
        smtp.send_message(msg=message)

Отправка HTML-письма.

Теперь обновим код выше для отправки HTML-письма.

Шаблон письма.

В корне проекта создадим директорию templates, а в ней файл confirmation_email.html.

В этом файле пропишем необходимый шаблон или простейший из примера ниже. Учтите, что многие почтовые клиенты не поддерживают или с опаской относятся к загружаемым со стороны стилям, поэтому стили лучше прописывать непосредственно в HTML-файле.

<!DOCTYPE html>  
<html lang="ru">  
<head>  
    <title>Подтверждение регистрации</title>  
</head>  
<body>  
<h1>Подтверждение регистрации</h1>  
<p>Для подтверждения регистрации перейдите по ссылке:</p>  
<a href="{{ confirmation_url }}">Подтвердить регистрацию</a>  
</body>  
</html>

Изменение задачи.

Чтобы отправлять HTML-письмо, необходимо изменить функцию send_confirmation_email. Переменная text больше не используется. Вместо неё теперь применяем три переменные:

  1. templates — В этой переменной создаём экземпляр Jinja2Templates, в который передаём параметр directory с указанием пути, по которому Jinja2 будет искать файлы шаблонов.
  2. template — В этой переменной загружается необходимый файл шаблона. Для этого у переменной templates вызывается метод .get_template, в который передаётся аргумент name — имя файла шаблона (например, "confirmation_email.html").
  3. html_content — Здесь получаем обработанный шаблон в виде HTML-строки. Для этого у переменной template вызывается метод .render, куда передаются ключевые аргументы или словарь с данными, которые должны быть доступны в шаблоне (аналогично тому, как это работает в Django).

Также изменяется способ задания содержимого письма. Вместо метода message.set_content теперь используется message.add_alternative, в который передаётся позиционный аргумент html_content и ключевой параметр subtype, указывающий, что содержимое — это HTML.

Всё, больше изменений не потребуется.

from starlette.templating import Jinja2Templates  


@shared_task  
def send_confirmation_email(to_email: str, token: str) -> None:  
    confirmation_url = f"{settings.frontend_url}/auth/register_confirm?token={token}"  

    templates = Jinja2Templates(directory=settings.templates_dir)  
    template = templates.get_template(name="confirmation_email.html")  
    html_content = template.render(confirmation_url=confirmation_url)  

    message = EmailMessage()  
    message.add_alternative(html_content, subtype="html")  
    message["From"] = settings.email_settings.email_username  
    message["To"] = to_email  
    message["Subject"] = "Подтверждение регистрации"  

    with smtplib.SMTP_SSL(host=settings.email_settings.email_host, port=settings.email_settings.email_port) as smtp:  
        smtp.login(  
            user=settings.email_settings.email_username,  
            password=settings.email_settings.email_password.get_secret_value(),  
        )  
        smtp.send_message(msg=message)

Выполнение задачи в пользовательском сервисе.

В нашем классе UserService в файле services.py мы просто регистрировали пользователя и возвращали его данные. В модели мы сразу предусмотрели поле is_active, определяющее активен пользователь или нет, равное False и is_verified определяющее подтверждена почта или нет, равное False. Осталось только добавить отправку письма.

Сперва добавим новое поле self.serializer в класс UserService. В этом поле определим экземпляр класса URLSafeTimedSerializer из библиотеки itsdangerous. Внутрь в аргумент secret_key передадим значение поля secret_key из класса Settings.

Коротко о URLSafeTimedSerializer - это класс, предназначенный для создания и проверки подписанных токенов, которые безопасны для включения в URL-адреса. Поддерживает создание и расшифровку токена на основе времени сервера, и контроль времени жизни токена.

В методе register_user, там где мы в самом конце возвращаем полученный объект пользователя, заменяем return на переменную user_data.

Ниже прописываем переменную confirmation_token, в которой обращаясь к полю self.serializer вызываем метод .dumps для создания нового токена. В метод передаём email пользователя для запечатывания в токен.

Далее вызываем метод .delay у функции send_confirmation_email из файла tasks.py, в него передаём два аргумента: to_email и token.

После всего этого делаем возврат значения переменной user_data.

Готово. Теперь, если вы запустите Celery и FastAPI, и попробуете зарегистрироваться, то получите письмо на электронную почту. Осталось только написать маршрут для подтверждения почты.

Обновлённый метод

async def register_user(self, user: RegisterUser) -> UserReturnData:  
    hashed_password = await self.handler.get_password_hash(user.password)  

    new_user = CreateUser(email=user.email, hashed_password=hashed_password)  

    user_data = await self.manager.create_user(user=new_user)  

    confirmation_token = self.serializer.dumps(user_data.email)  
    send_confirmation_email.delay(to_email=user_data.email, token=confirmation_token)  

    return user_data

 

Маршрут подтверждения регистрации.

После того, как пользователь получит письмо со ссылкой, необходимо обработать переход по ссылке и активацию учётной записи.

Сохранение изменений в базе данных.

Начнём мы не с маршрута и дальше, а наоборот с конца. Сперва напишем метод в пользовательском менеджере для активации учётной записи.

Откроем файл managers.py и в классе UserManager создадим новый метод confirm_user, принимающий self и email.

При помощи async with открываем контекстный менеджер для создания сессии БД.

Внутри менеджера прописываем переменную query, в которой сформируем запрос:

  1. Вызываем функцию update, сообщая ORM, что мы хотим обновить запись. В аргументы передаём поле self.model.
  2. Вызываем метод .where, указывая, что нам нужно обновить записи соответствующие определённому условию. В аргументы передаём условие, что self.model.email - электронная почта пользователя равна переданной в метод электронной почте.
  3. Вызываем метод .values, в который передаём ключевыми аргументами имена полей и значения для обновления. В аргументах указываем поля is_verified и is_active равные True.

Затем обращаясь к методу .execute у переменной session, передаём запрос на выполнение.
После чего, обращаясь к методу .commit у переменной session сохраняем изменения в базе данных.

Код метода:

from sqlalchemy import update

async def confirm_user(self, email: str) -> None:  
    async with self.db.db_session() as session:  
        query = (  
            update(self.model)  
            .where(self.model.email == email)  
            .values(is_verified=True, is_active=True)  
        )  
        await session.execute(query)  
        await session.commit()

Вызов сохранения в пользовательском сервисе.

Откроем файл services.py и в классе UserService создадим метод confirm_user, принимающий self и token.

Внутри функции создаём try-except-блок.
В блоке try определяем переменную email, в которой обращаясь к методу .loads у поля self.serializer, расшифровываем полученный токен. Внутрь передаём два аргумента:

  • token - Позиционный аргумент с токеном.
  • max_age - Ключевой аргумент в котором определяется допустимое время жизни токена, т.е. если токен был создан 10 минут назад, а допустимое время жизни 5 минут, то токен будет считаться просроченным и выдаст ошибку.

В блоке except "отлавливаем" класс ошибки BadSignature.
Внутри блока except при срабатывании поднимаем (raise) ошибку HTTPException, передав в неё статус-код ошибки, в нашем случае 400 и сообщение об ошибке.

После этого, если ошибки не было, вызываем созданный ранее метод .confirm_user у поля self.manager.

Код метода

from fastapi import HTTPException
from itsdangerous import BadSignature


async def confirm_user(self, token: str) -> None:  
    try:  
        email = self.serializer.loads(token, max_age=3600)  
    except BadSignature:  
        raise HTTPException(  
            status_code=400, detail="Неверный или просроченный токен"  
        )  

    await self.manager.confirm_user(email=email)

Маршрут подтверждения.

Откроем файл routes.py и создадим новую функцию confirm_registration.

Функция принимает два аргумента:

  1. token - Созданный ранее токен для подтверждения почты.
  2. service - Объект типа UserService, используем функцию Depends как в маршруте регистрации.

Сразу оборачиваем функцию в декоратор @auth_router.get, сообщая, что он будет обрабатывать GET-запросы. В декоратор передаём два аргумента:

  1. path - Путь который будет обрабатывать маршрут. В нашем случае это "/register_confirm".
  2. status_code - Возвращаемый статус-код при успешном выполнении. Установим стандартный status.HTTP_200_OK.

Вызываем у аргумента service наш метод confirm_user, передав в него полученный token.

После чего возвращаем сообщение об успешном подтверждении почты, если не сработало исключение.

Код маршрута:

@auth_router.get(path="/register_confirm", status_code=status.HTTP_200_OK)  
async def confirm_registration(token: str, service: UserService = Depends(UserService)) -> dict[str, str]:  
    await service.confirm_user(token=token)  
    return {"message": "Электронная почта подтверждена"}

 

Тестирование

Для тестирования, необходимо запустить FastAPI и Celery одновременно в двух терминалах.

В первом терминале запускаем FastAPI:

poetry run app

Во втором терминале запускаем Celery:

celery -A lkeep.core.celery_app worker -l INFO

Перейдём в Swagger-документацию, открыв в браузере http://127.0.0.1:8000/docs

Раскроем маршрут регистрации:

 

И нажмём Try it out:

 

В появившемся поле впишите вашу почту и случайный пароль. Затем нажмите Execute.

Прокрутив немного ниже, в блоке Server response, если всё настроенно корректно будет информация о созданном пользователе:

 

Перейдя на почту, увидим наше письмо:

Текстовое:

 

HTML:

 

После перехода по ссылке, получим ответ, что электронная почта подтверждена:

 

 

Заключение.

Подтверждение регистрации по электронной почте — это не просто дополнительный шаг в процессе регистрации, это важный механизм, повышающий безопасность и качество работы сервиса. Зарегистрированный пользователь — это хорошо, но подтверждённый по почте пользователь даёт возможность убедиться в реальности учетной записи, что значительно снижает риск массовой регистрации ботами. Такие автоматизированные регистрации могут не только исказить статистику, но и негативно повлиять на работу проекта, нарушая целостность пользовательской базы.

В нашей реализации мы использовали целый набор инструментов, каждый из которых играет свою уникальную роль:

  • Celery и Redis обеспечивают асинхронную обработку задач, что позволяет отправлять почту в фоновом режиме и не замедлять работу основного приложения.
  • smtplib отвечает за надежное подключение к почтовому серверу и доставку писем, а jinja2 делает возможным создание красивых и удобочитаемых HTML-шаблонов, что улучшает визуальное восприятие сообщений.
  • itsdangerous гарантирует безопасность токенов для подтверждения регистрации, защищая данные пользователей и предотвращая их подделку.

 

Репозиторий проекта в GitHub: https://github.com/proDreams/lkeep 
Репозиторий проекта в "GIT на салфетке": https://git.pressanybutton.ru/proDream/lkeep
 

Если заметили ошибку — напишите в Telegram автора.
Аватар автора

Автор

Иван Ашихмин

Программист, фрилансер и автор гайдов. Занимаюсь разработкой ботов, сайтов и не только.

Telegram GitHub Сайт Поддержать

Комментарии

Оставить комментарий

Войдите, чтобы оставить комментарий.

Комментариев пока нет.