Cat

FastAPI 5. Приложение аутентификации и Pydantic схемы

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

Сервис на FastAPI proDream 19 Декабрь 2024 Просмотров: 198

Весь наш проект будет состоять из набора приложений, которые можно сравнить с подходом в Django. Каждое приложение будет выполнять строго определённую задачу. Например, первое приложение, которое мы создадим, будет называться auth, и оно займётся функционалом, связанным с пользователями и системой аутентификации.

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


 

Структура

Чтобы проект оставался структурированным и удобным для поддержки, начнём с подготовки структуры для приложений.

  1. В корневом пакете проекта lkeep создадим новый пакет apps. Он будет служить контейнером для всех приложений проекта.
  2. Внутри пакета 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:

  1. Валидация данных 
    Pydantic проверяет входящие данные на соответствие заданным требованиям. Если, например, пользователь отправил строку вместо числа или пропустил обязательное поле, FastAPI автоматически вернёт сообщение об ошибке с кодом состояния 400 Bad Request.
  2. Сериализация данных 
    При отправке ответа Pydantic преобразует Python-объекты (например, словари или списки) в формат JSON, который клиент может понять.
  3. Создание объектов 
    С помощью Pydantic можно создавать строгие классы для представления данных. Эти классы гарантируют, что данные внутри объекта соответствуют определённым типам и ограничениям.
  4. Автоматическое генерирование документации 
    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. В этом файле будут описаны схемы, необходимые для работы с пользователем.

Для начала мы реализуем схему регистрации пользователя. Нам понадобятся пять схем:

  1. GetUserByID
    • Содержит единственное поле id с типом данных uuid.UUID | str.
    • Используется для получения пользователя по его уникальному идентификатору.
    • Наследуется от BaseModel.
  2. GetUserByEmail
    • Содержит единственное поле email с типом данных EmailStr.
    • Используется для получения пользователя по электронной почте.
    • Наследуется от BaseModel.
  3. RegisterUser
    • Содержит поле password с типом данных str.
    • Используется при регистрации пользователя.
    • Наследуется от GetUserByEmail, чтобы переиспользовать поле email.
  4. CreateUser
    • Содержит поле hashed_password с типом данных str.
    • Используется для создания записи пользователя в базе данных.
    • Наследуется от GetUserByEmail, чтобы переиспользовать поле email.
  5. 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, должны быть классами или функциями без вызова (без круглых скобок).

Внутри конструктора мы:

  1. Сохраняем переданный объект db в поле self.db.
  2. Указываем модель пользователя в поле 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

 

Пояснения

  1. Аргумент db:
    • Это зависимость, предоставляющая доступ к базе данных.
    • Использование Depends позволяет FastAPI автоматически создавать экземпляр зависимости при вызове методов, где используется менеджер.
  2. Поля класса:
    • self.db: хранит объект базы данных для выполнения операций (например, запросов).
    • self.model: указывает на модель, с которой работает менеджер (например, User).

 

Что такое Depends?

Depends — это функция из FastAPI, которая используется для внедрения зависимостей в ваше приложение. Зависимости могут быть любыми объектами или ресурсами, которые нужны для работы функции или класса, например:

  • Сессия базы данных.
  • Пользователь, прошедший аутентификацию.
  • Настройки приложения.

FastAPI автоматически создаёт и передаёт зависимости туда, где они указаны, избавляя вас от лишнего шаблонного кода.

 

Как это работает?

Depends действует как связующее звено между описанием того, что требуется функции или классу, и механизмом создания или получения этих ресурсов.

Когда FastAPI обнаруживает зависимость (например, через Depends(DBDependency)), оно:

  1. Создаёт или берёт готовый экземпляр ресурса.
  2. Автоматически передаёт его в функцию или класс.

Это помогает избежать ручного управления подключениями и делает код понятным и лаконичным.

 

Пример использования

В нашем коде мы применяем 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.

 

Почему это полезно?

  1. Меньше шаблонного кода: Не нужно вручную создавать или передавать зависимости.
  2. Управление жизненным циклом: FastAPI автоматически создаёт и закрывает ресурсы (например, сессии базы данных).
  3. Повторное использование: Один и тот же ресурс можно использовать в разных функциях или классах.
  4. Гибкость: Легко заменить зависимости на моки для тестирования.

 

Метод create_user

Метод create_user отвечает за создание нового пользователя в базе данных. Мы будем использовать асинхронный метод, который принимает два аргумента:

  • self — экземпляр класса UserManager.
  • user — объект класса схемы CreateUser, содержащий данные для создания пользователя (email и hashed_password).

Метод вернёт объект, соответствующий схеме UserReturnData, содержащей информацию о новом пользователе.

 

Разбор по шагам:

  1. Открываем сессию базы данных: 
    В теле функции используем поле db_session объекта self.db (созданного ранее через Depends) и открываем асинхронный контекстный менеджер:
    python async with self.db.db_session as session: 
    Это позволяет автоматически управлять подключением к базе данных и закрывать его после завершения операции.
  2. Формируем запрос: Создаём переменную 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)
  3. Обрабатываем возможные ошибки: 
    Используем блок 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.")
  4. Применяем изменения: 
    После успешного выполнения запроса вызываем await session.commit(), чтобы сохранить изменения в базе данных.
  5. Получаем данные о пользователе: 
    Сохраняем объект созданного пользователя в переменной user_data, используя метод scalar_one() для получения одного результата:
    python user_data = await result.scalar_one()
  6. Возвращаем данные: 
    Преобразуем объект пользователя в 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__)

 

Ключевые моменты:

  1. Асинхронность: Метод полностью асинхронный, что позволяет эффективно работать с базой данных.
  2. Работа с Pydantic: Используем Pydantic для валидации входных данных (CreateUser) и форматирования выходных данных (UserReturnData).
  3. Обработка ошибок: Предусмотрена защита от дублирования данных через перехват исключений.
  4. Контекстный менеджер: Управление сессией базы данных реализовано через асинхронный контекстный менеджер для упрощения и безопасности.

 

Полный код файла:

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, чтобы он был доступен в приложении.

  1. Откройте файл 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"
    )
  1. Затем откройте .env в корне проекта и добавьте строку с вашим секретным ключом:
    SECRET_KEY=f4b5f7a40d0c4c1b9db0d34913c4e13c9f51f5c2b9d0309059efbdf3fe332b2c

Теперь секретный ключ доступен через объект settings.secret_key.

 

Установка библиотеки passlib для работы с паролями

Для безопасного хранения паролей мы будем использовать библиотеку passlib. Она поддерживает множество алгоритмов хэширования, включая bcrypt, который является стандартным и надёжным выбором.

  1. Установите библиотеку через poetry с указанием модуля bcrypt:
poetry add passlib 'bcrypt4.0.1'
  1. Почему указывается версия bcrypt? 
    Новые версии bcrypt могут вызвать ошибки при использовании с passlib, так как изменения в API библиотеки ещё не учтены в passlib. Указание версии bcrypt4.0.1 помогает избежать подобных проблем. Возможно, в будущем эта несовместимость будет устранена.

 

Класс AuthHandler

Теперь давайте перейдём к классу AuthHandler, который будет заниматься аутентификацией и безопасностью в нашем приложении. В классе определим два поля:

  • secret — секретный ключ, который будет использоваться для шифрования. Мы получим его через метод .get_secret_value(), который возвращает саму строку секрета.
  • pwd_context — экземпляр класса CryptContext из библиотеки passlib. Это контекст для работы с хэшированием паролей, в котором мы укажем два параметра:
    1. schemes=["bcrypt"] — схему хэширования пароля, в данном случае это bcrypt, который является одним из самых безопасных методов. Он использует соль, что значительно усложняет атаки методом подбора пароля.
    2. 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

Автор

    Нет комментариев

    Реклама