FastAPI 5. Приложение аутентификации и Pydantic схемы
В этом посте мы напишем Pydantic-схемы для данных пользователя, создадим класс пользовательского менеджера для работы с БД, а также пользовательский обработчик для работы в паролями и другими видами токенов.
Реклама
Весь наш проект будет состоять из набора приложений, которые можно сравнить с подходом в Django. Каждое приложение будет выполнять строго определённую задачу. Например, первое приложение, которое мы создадим, будет называться auth
, и оно займётся функционалом, связанным с пользователями и системой аутентификации.
В этом посте мы создадим основу приложения аутентификации, начав с реализации регистрации пользователя. В дальнейшем мы добавим другие методы, такие как авторизация, восстановление пароля и управление пользователями.
Структура
Чтобы проект оставался структурированным и удобным для поддержки, начнём с подготовки структуры для приложений.
- В корневом пакете проекта
lkeep
создадим новый пакетapps
. Он будет служить контейнером для всех приложений проекта. - Внутри пакета
apps
создадим ещё один пакет —auth
. Этот пакет будет содержать все файлы, связанные исключительно с системой аутентификации: регистрацию, авторизацию, работу с токенами и другие связанные функции.
Пример структуры:
lkeep/
├── apps/
│ ├── auth/
│ │ ├── __init__.py
│ │ ├── handlers.py
│ │ ├── managers.py
│ │ ├── routes.py
│ │ ├── schemas.py
│ │ ├── services.py
│ │ └── utils.py
Такая структура помогает изолировать функционал отдельных приложений и делает код более читабельным и удобным для тестирования.
Pydantic для валидации
Для работы с API нам понадобятся маршруты, а для обработки входящих запросов и формирования ответов — схемы.
Одной из ключевых особенностей FastAPI является использование библиотеки Pydantic. Она отвечает за валидацию данных, которые поступают в запросах, и сериализацию данных, которые возвращаются в ответах.
Основные возможности Pydantic в FastAPI:
- Валидация данных
Pydantic проверяет входящие данные на соответствие заданным требованиям. Если, например, пользователь отправил строку вместо числа или пропустил обязательное поле, FastAPI автоматически вернёт сообщение об ошибке с кодом состояния400 Bad Request
. - Сериализация данных
При отправке ответа Pydantic преобразует Python-объекты (например, словари или списки) в формат JSON, который клиент может понять. - Создание объектов
С помощью Pydantic можно создавать строгие классы для представления данных. Эти классы гарантируют, что данные внутри объекта соответствуют определённым типам и ограничениям. - Автоматическое генерирование документации
FastAPI использует Pydantic для генерации документации Swagger и OpenAPI. Поля, типы данных и ограничения из схем автоматически отображаются в документации, что упрощает использование API.
Схемы в FastAPI
Схемы (Pydantic-модели) — это классы, которые описывают структуру данных. Они содержат:
- Типы полей (например,
str
,int
,list[str]
), - Значения по умолчанию,
- Валидацию (например, максимальная длина строки, проверка формата e-mail),
- Дополнительные параметры (например, поле может быть необязательным).
Вот пример простой Pydantic-схемы:
from pydantic import BaseModel, EmailStr
class UserRegistration(BaseModel):
username: str
email: EmailStr
password: str
Этот класс:
- Проверяет, что
username
иpassword
— это строки, - Убедится, что
email
имеет корректный формат e-mail.
Схемы в FastAPI можно сравнить с моделями в Django, однако они используются для валидации данных и не связаны напрямую с базой данных.
Схемы для пользователя
Для описания структур данных создадим файл schemas.py
в директории приложения auth
. В этом файле будут описаны схемы, необходимые для работы с пользователем.
Для начала мы реализуем схему регистрации пользователя. Нам понадобятся пять схем:
GetUserByID
- Содержит единственное поле
id
с типом данныхuuid.UUID | str
. - Используется для получения пользователя по его уникальному идентификатору.
- Наследуется от
BaseModel
.
- Содержит единственное поле
GetUserByEmail
- Содержит единственное поле
email
с типом данныхEmailStr
. - Используется для получения пользователя по электронной почте.
- Наследуется от
BaseModel
.
- Содержит единственное поле
RegisterUser
- Содержит поле
password
с типом данныхstr
. - Используется при регистрации пользователя.
- Наследуется от
GetUserByEmail
, чтобы переиспользовать полеemail
.
- Содержит поле
CreateUser
- Содержит поле
hashed_password
с типом данныхstr
. - Используется для создания записи пользователя в базе данных.
- Наследуется от
GetUserByEmail
, чтобы переиспользовать полеemail
.
- Содержит поле
UserReturnData
- Содержит следующие поля:
is_active
— активность пользователя,is_verified
— подтверждён ли пользователь,is_superuser
— является ли пользователь суперпользователем,created_at
— дата создания,updated_at
— дата обновления.
- Наследуется от
GetUserByID
иGetUserByEmail
, чтобы переиспользовать поляid
иemail
.
- Содержит следующие поля:
Преимущества наследования
Использование наследования позволяет избежать дублирования кода. Вместо того чтобы каждый раз заново описывать одинаковые поля, мы переиспользуем их из базовых классов. Это особенно удобно, когда схема требует добавления новых полей — изменения нужно вносить только в одном месте.
О поле EmailStr
EmailStr
— это специальный тип данных из Pydantic, который проверяет корректность адресов электронной почты. Однако для его работы требуется установить библиотеку email-validator
. Установить её можно одной из следующих команд:
poetry add "pydantic[email]"
или
poetry add email-validator
Пример кода
import uuid
import datetime
from pydantic import BaseModel, EmailStr, Field
class GetUserByID(BaseModel):
id: uuid.UUID | str
class GetUserByEmail(BaseModel):
email: EmailStr
class RegisterUser(GetUserByEmail):
password: str
class CreateUser(GetUserByEmail):
hashed_password: str
class UserReturnData(GetUserByID, GetUserByEmail):
is_active: bool
is_verified: bool
is_superuser: bool
created_at: datetime.datetime
updated_at: datetime.datetime
Объяснение структуры
- Поле
id
используется для уникальной идентификации пользователей. Оно принимает значение UUID, что предотвращает пересечения идентификаторов. - Поле
email
проверяется на соответствие формату электронной почты. - Поле
password
предназначено только для ввода пароля пользователем, аhashed_password
используется для хранения зашифрованного пароля в базе данных. - Поля
is_active
,is_verified
и другие помогут в управлении статусами пользователей.
Схемы — это связующее звено между клиентом и сервером. Они позволяют разработчику описывать данные структурировано и минимизируют ошибки на уровне API.
Пользовательский менеджер
Пользовательский менеджер — это класс, который предоставляет удобный интерфейс для управления бизнес-логикой, связанной с данными пользователя.
Зачем нужен пользовательский менеджер?
- Сокрытие сложности: Этот класс изолирует операции с базой данных за простым и понятным API, что делает код читаемым и удобным для использования.
- Повторное использование: Выделение общей логики в отдельный класс позволяет избежать дублирования кода.
- Лёгкость модификации: При необходимости изменения логики работы с пользователями потребуется обновить только методы менеджера, что упрощает поддержку и тестирование.
Создание пользовательского менеджера
В пакете auth
создадим файл managers.py
. Здесь мы опишем класс UserManager
.
Конструктор класса
Начнём с определения конструктора. В Python конструктор задаётся с помощью специального метода __init__
.
- Конструктор принимает параметр
db
: зависимость, отвечающая за работу с базой данных (это экземпляр классаDBDependency
). - Для аргумента
db
мы используем значение по умолчанию —Depends(DBDependency)
.
Важно: При использовании FastAPI зависимости, передаваемые через
Depends
, должны быть классами или функциями без вызова (без круглых скобок).
Внутри конструктора мы:
- Сохраняем переданный объект
db
в полеself.db
. - Указываем модель пользователя в поле
self.model
.
Код конструктора
from fastapi import Depends
from app.dependencies import DBDependency # Зависимость для работы с БД
from app.models import User # Модель пользователя
class UserManager:
def __init__(self, db: DBDependency = Depends(DBDependency)) -> None:
self.db = db
self.model = User
Пояснения
- Аргумент
db
:- Это зависимость, предоставляющая доступ к базе данных.
- Использование
Depends
позволяет FastAPI автоматически создавать экземпляр зависимости при вызове методов, где используется менеджер.
- Поля класса:
self.db
: хранит объект базы данных для выполнения операций (например, запросов).self.model
: указывает на модель, с которой работает менеджер (например,User
).
Что такое Depends
?
Depends
— это функция из FastAPI, которая используется для внедрения зависимостей в ваше приложение. Зависимости могут быть любыми объектами или ресурсами, которые нужны для работы функции или класса, например:
- Сессия базы данных.
- Пользователь, прошедший аутентификацию.
- Настройки приложения.
FastAPI автоматически создаёт и передаёт зависимости туда, где они указаны, избавляя вас от лишнего шаблонного кода.
Как это работает?
Depends
действует как связующее звено между описанием того, что требуется функции или классу, и механизмом создания или получения этих ресурсов.
Когда FastAPI обнаруживает зависимость (например, через Depends(DBDependency)
), оно:
- Создаёт или берёт готовый экземпляр ресурса.
- Автоматически передаёт его в функцию или класс.
Это помогает избежать ручного управления подключениями и делает код понятным и лаконичным.
Пример использования
В нашем коде мы применяем Depends
для автоматической передачи объекта базы данных в конструктор класса UserManager
:
class UserManager:
def __init__(self, db: DBDependency = Depends(DBDependency)) -> None:
self.db = db
self.model = User
db: DBDependency = Depends(DBDependency)
:- Здесь
Depends
сообщает FastAPI, что нам нужен объект, предоставляемый классомDBDependency
. - FastAPI самостоятельно создаёт этот объект и передаёт его в конструктор
UserManager
.
- Здесь
Почему это полезно?
- Меньше шаблонного кода: Не нужно вручную создавать или передавать зависимости.
- Управление жизненным циклом: FastAPI автоматически создаёт и закрывает ресурсы (например, сессии базы данных).
- Повторное использование: Один и тот же ресурс можно использовать в разных функциях или классах.
- Гибкость: Легко заменить зависимости на моки для тестирования.
Метод create_user
Метод create_user
отвечает за создание нового пользователя в базе данных. Мы будем использовать асинхронный метод, который принимает два аргумента:
self
— экземпляр классаUserManager
.user
— объект класса схемыCreateUser
, содержащий данные для создания пользователя (email
иhashed_password
).
Метод вернёт объект, соответствующий схеме UserReturnData
, содержащей информацию о новом пользователе.
Разбор по шагам:
- Открываем сессию базы данных:
В теле функции используем полеdb_session
объектаself.db
(созданного ранее черезDepends
) и открываем асинхронный контекстный менеджер:python async with self.db.db_session as session:
Это позволяет автоматически управлять подключением к базе данных и закрывать его после завершения операции. - Формируем запрос: Создаём переменную
query
, в которой формируем запрос на добавление данных:insert(self.model)
— вставка данных в таблицу, связанной с модельюself.model
(модель пользователя)..values(**user.model_dump())
— распаковываем данные из объекта схемыCreateUser
с помощью метода.model_dump()
, чтобы передать их в запрос в виде словаря..returning(self.model)
— указываем, что запрос должен вернуть объект модели после выполнения.
Итоговый код:python query = insert(self.model).values(**user.model_dump()).returning(self.model)
- Обрабатываем возможные ошибки:
Используем блокtry-except
, чтобы "отловить" исключениеIntegrityError
, которое возникает, если в базе уже существует пользователь с такимemail
. В случае ошибки выбрасываемHTTPException
с кодом400 Bad Request
:python try: result = await session.execute(query) except IntegrityError: raise HTTPException(status_code=400, detail="User already exists.")
- Применяем изменения:
После успешного выполнения запроса вызываемawait session.commit()
, чтобы сохранить изменения в базе данных. - Получаем данные о пользователе:
Сохраняем объект созданного пользователя в переменнойuser_data
, используя методscalar_one()
для получения одного результата:python user_data = await result.scalar_one()
- Возвращаем данные:
Преобразуем объект пользователя в Pydantic-схемуUserReturnData
, распаковывая его атрибуты через__dict__
:python return UserReturnData(**user_data.__dict__)
Финальный код метода:
async def create_user(self, user: CreateUser) -> UserReturnData:
async with self.db.db_session as session:
query = insert(self.model).values(**user.model_dump()).returning(self.model)
try:
result = await session.execute(query)
except IntegrityError:
raise HTTPException(status_code=400, detail="User already exists.")
await session.commit()
user_data = await result.scalar_one()
return UserReturnData(**user_data.__dict__)
Ключевые моменты:
- Асинхронность: Метод полностью асинхронный, что позволяет эффективно работать с базой данных.
- Работа с Pydantic: Используем Pydantic для валидации входных данных (
CreateUser
) и форматирования выходных данных (UserReturnData
). - Обработка ошибок: Предусмотрена защита от дублирования данных через перехват исключений.
- Контекстный менеджер: Управление сессией базы данных реализовано через асинхронный контекстный менеджер для упрощения и безопасности.
Полный код файла:
from fastapi import Depends, HTTPException
from sqlalchemy import insert
from sqlalchemy.exc import IntegrityError
from lkeep.apps.auth.schemas import CreateUser, UserReturnData
from lkeep.core.core_dependency.db_dependency import DBDependency
from lkeep.database.models import User
class UserManager:
def __init__(self, model: type[User] = User,db: DBDependency = Depends(DBDependency)) -> None:
self.db = db
self.model = model
async def create_user(self, user: CreateUser) -> UserReturnData:
async with self.db.db_session as session:
query = insert(self.model).values(**user.model_dump()).returning(self.model)
try:
result = await session.execute(query)
except IntegrityError:
raise HTTPException(status_code=400, detail="User already exists.")
await session.commit()
user_data = await result.scalar_one()
return UserReturnData(**user_data.__dict__)
Обработчик аутентификации
Для выполнения операций, связанных с аутентификацией и безопасностью, таких как расшифровка токенов или шифрование паролей, нам понадобится отдельный обработчик.
В пакете auth
создадим файл handlers.py
и определим в нём класс AuthHandler
.
Для реализации обработчика потребуется установить дополнительную библиотеку и добавить новый параметр в конфигурацию проекта.
Секретный ключ и библиотека passlib
Обработчику аутентификации, который мы будем писать далее, нужен секретный ключ. Он используется для подписания токенов, таких как токен восстановления пароля, а также для их проверки.
Генерация секретного ключа
Секретный ключ — это строка, содержащая случайный набор символов, которая обеспечивает безопасность операций с токенами. Вы можете создать его вручную или сгенерировать с помощью Python.
Шаги для генерации ключа:
Откройте Python Console в вашей IDE или в терминале выполните команду:
python
Используйте следующий код для генерации ключа с помощью модуля secrets
:
# Импортируем модуль secrets
import secrets
# Генерируем строку
print(secrets.token_hex(32)) # 32 — это длина ключа
В результате вы получите строку, например:
f4b5f7a40d0c4c1b9db0d34913c4e13c9f51f5c2b9d0309059efbdf3fe332b2c
Совет: Сохраняйте этот ключ в секрете и не публикуйте его в репозиториях.
Чтобы выйти из консоли Python, выполните:
exit()
Добавление секретного ключа в конфигурацию
После генерации секретного ключа добавьте его в файл настроек settings.py
, чтобы он был доступен в приложении.
- Откройте файл
settings.py
и обновите классSettings
, добавив полеsecret_key
. Используйте типSecretStr
из Pydantic для безопасного хранения секретных данных. Также укажите, что настройки будут считываться из файла.env
.
from pydantic import BaseSettings, SecretStr, SettingsConfigDict
class Settings(BaseSettings):
db_settings: DBSettings = DBSettings()
secret_key: SecretStr
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf8",
extra="ignore"
)
- Затем откройте
.env
в корне проекта и добавьте строку с вашим секретным ключом:SECRET_KEY=f4b5f7a40d0c4c1b9db0d34913c4e13c9f51f5c2b9d0309059efbdf3fe332b2c
Теперь секретный ключ доступен через объект settings.secret_key
.
Установка библиотеки passlib
для работы с паролями
Для безопасного хранения паролей мы будем использовать библиотеку passlib
. Она поддерживает множество алгоритмов хэширования, включая bcrypt
, который является стандартным и надёжным выбором.
- Установите библиотеку через
poetry
с указанием модуляbcrypt
:
poetry add passlib 'bcrypt4.0.1'
- Почему указывается версия
bcrypt
?
Новые версииbcrypt
могут вызвать ошибки при использовании сpasslib
, так как изменения в API библиотеки ещё не учтены вpasslib
. Указание версииbcrypt4.0.1
помогает избежать подобных проблем. Возможно, в будущем эта несовместимость будет устранена.
Класс AuthHandler
Теперь давайте перейдём к классу AuthHandler
, который будет заниматься аутентификацией и безопасностью в нашем приложении. В классе определим два поля:
secret
— секретный ключ, который будет использоваться для шифрования. Мы получим его через метод.get_secret_value()
, который возвращает саму строку секрета.pwd_context
— экземпляр классаCryptContext
из библиотекиpasslib
. Это контекст для работы с хэшированием паролей, в котором мы укажем два параметра:schemes=["bcrypt"]
— схему хэширования пароля, в данном случае этоbcrypt
, который является одним из самых безопасных методов. Он использует соль, что значительно усложняет атаки методом подбора пароля.deprecated="auto"
— данный параметр указывает, что если схема хэширования станет устаревшей,passlib
автоматически перейдёт на более современную.
Кроме того, создадим асинхронный метод get_password_hash
, который будет принимать пароль и возвращать его хэшированное значение в строковом формате.
Код класса AuthHandler
:
from passlib.context import CryptContext
from lkeep.core.settings import settings
class AuthHandler:
secret = settings.secret_key.get_secret_value()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
async def get_password_hash(self, password: str) -> str:
return self.pwd_context.hash(password)
Заключение
В этом посте мы рассмотрели несколько важных шагов для создания защищённого приложения на FastAPI. Мы разработали Pydantic-модели, создали менеджера пользователей и обработчик аутентификации. Также реализовали метод для безопасного хэширования паролей.
Эти базовые принципы, которые мы рассмотрели, являются основой для построения надёжных и безопасных приложений. В будущих постах мы продолжим развивать эту тему, рассматривая более сложные и масштабируемые решения для ваших сервисов.
Репозиторий проекта в GitHub: https://github.com/proDreams/lkeep
Репозиторий проекта в "GIT на салфетке": https://git.pressanybutton.ru/proDream/lkeep
Все статьи