Создание приложений на FastAPI. Часть первая: Введение и подготовка

FastAPI — это современный, быстрый (high-performance) веб-фреймворк для создания API на языке Python, основанный на стандартных спецификациях OpenAPI и JSON Schema. С момента своего появления, FastAPI стал чрезвычайно популярным среди разработчиков, благодаря своему уникальному сочетанию простоты использования, высокой производительности и расширяемости. Основатель и главный разработчик FastAPI, Себастьян Рамирес, сумел создать инструмент, который не только облегчает процесс разработки, но и повышает его качество, делая код более чистым и понятным.

Основные преимущества FastAPI

1. Скорость работы: Одной из ключевых особенностей FastAPI является его невероятная скорость работы. Это достигается благодаря использованию асинхронного программирования (asyncio) и Uvicorn — высокопроизводительного ASGI-сервера, который лежит в основе фреймворка. Согласно различным тестам производительности, FastAPI способен обрабатывать сотни тысяч запросов в секунду, что ставит его на один уровень с такими известными фреймворками, как Node.js и Go. Это делает его идеальным выбором для создания высоконагруженных систем и API, которые требуют минимальной задержки.

2. Простота и интуитивность: FastAPI предоставляет разработчикам инструменты, которые делают создание API интуитивно понятным процессом. Использование аннотаций типов Python позволяет автоматически генерировать схемы данных, документацию и валидацию входных данных. Это уменьшает количество кода, который нужно писать вручную, и минимизирует вероятность ошибок. Более того, встроенная документация API, доступная через Swagger UI и Redoc, позволяет в реальном времени видеть и тестировать все доступные эндпоинты, что упрощает процесс разработки и интеграции.

3. Асинхронность и поддержка современного Python: Асинхронность — это одна из главных причин, по которой разработчики выбирают FastAPI для своих проектов. Встроенная поддержка асинхронных операций позволяет эффективно работать с большим количеством одновременных запросов, что особенно важно для приложений, работающих с I/O операциями, такими как работа с базами данных или внешними API. FastAPI использует стандартные механизмы Python для реализации асинхронности, такие как async и await, что делает код чистым и легко читаемым.

4. Поддержка стандартов OpenAPI и JSON Schema: FastAPI полностью поддерживает спецификации OpenAPI и JSON Schema, что позволяет автоматически генерировать документацию для API и схемы данных. Это значительно упрощает процесс интеграции с другими системами и создания клиентских приложений. Поддержка этих стандартов делает FastAPI особенно привлекательным для компаний, которые ценят стандартизацию и взаимосовместимость.

5. Совместимость и расширяемость: FastAPI построен на основе Starlette и Pydantic, двух мощных библиотек, которые отвечают за маршрутизацию, асинхронность и валидацию данных соответственно. Это позволяет легко интегрировать FastAPI с другими инструментами и библиотеками, такими как SQLAlchemy, Celery, и Redis. Кроме того, структура FastAPI очень гибкая, что позволяет легко расширять его функциональность и адаптировать под специфические потребности проекта.

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

FastAPI: Сердце приложения

FastAPI — это не просто основа нашего приложения, это его сердце. Как уже упоминалось, этот фреймворк был выбран за его высокую производительность, поддержку асинхронного программирования и автоматическую генерацию документации. Однако, важно подчеркнуть, что FastAPI также предлагает отличную интеграцию с другими библиотеками и инструментами, что делает его универсальным решением для разработки как простых, так и сложных API.

SQLAlchemy [asyncio]: Асинхронная работа с базой данных

Для работы с базами данных в проекте используется SQLAlchemy с поддержкой асинхронных операций (sqlalchemy[asyncio]). SQLAlchemy — это один из самых популярных ORM (Object-Relational Mapping) инструментов для Python, который предоставляет удобный интерфейс для работы с базами данных. В нашем случае мы используем его асинхронную версию, что позволяет эффективно управлять ресурсами и обеспечивать высокую производительность при работе с большим количеством одновременных запросов к базе данных.

Асинхронный подход к работе с базой данных особенно важен для приложений, работающих под высокой нагрузкой, так как позволяет избежать блокировок и задержек при обработке запросов, что, в свою очередь, положительно сказывается на общей производительности приложения.

Pydantic: Валидация и аннотации типов

Pydantic является фундаментальной библиотекой для FastAPI, так как он отвечает за валидацию данных и аннотации типов. В основе Pydantic лежит использование аннотаций типов Python для проверки данных и их приведения к нужному формату. Это не только упрощает процесс разработки, но и делает код более надежным и понятным.

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

Alembic: Миграции базы данных

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

Использование Alembic в связке с SQLAlchemy позволяет нам легко управлять версиями базы данных, откатывать изменения и поддерживать актуальное состояние схемы базы данных на всех этапах разработки.

Redis [asyncio]: Кэширование и хранение состояний

Redis — это высокопроизводительная база данных в памяти, которая используется для кэширования и хранения состояний. В нашем проекте Redis выполняет несколько ключевых ролей, включая кэширование результатов запросов, хранение сессий пользователей и управление очередями задач.

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

Celery и/или TaskIQ: Управление фоновыми задачами

Для выполнения фоновых задач в нашем приложении используются Celery и TaskIQ. Эти инструменты позволяют эффективно распределять и обрабатывать задачи, которые не требуют немедленного выполнения, что освобождает основные ресурсы приложения и повышает его отзывчивость.

TaskIQ — это библиотека, созданная специально для интеграции с FastAPI, что делает её идеальным выбором для управления фоновыми задачами в контексте нашего проекта. Celery, в свою очередь, является более зрелым и мощным инструментом, который поддерживает различные брокеры сообщений (RabbitMQ, Redis) и может быть использован для более сложных сценариев обработки задач в синхронном режиме, например генерация изображений.

FastAPI Users: Аутентификация и авторизация

FastAPI Users — это библиотека, предназначенная для быстрого и простого внедрения системы управления пользователями в проекты на FastAPI. Она поддерживает все основные функции, необходимые для работы с пользователями, включая регистрацию, вход, восстановление пароля и управление ролями.

В нашем проекте FastAPI Users играет ключевую роль в обеспечении безопасности и управлении доступом к различным частям приложения. Эта библиотека интегрируется с SQLAlchemy для хранения данных пользователей и поддерживает различные способы аутентификации, включая JWT (JSON Web Tokens).

FastAPI Pagination: Пагинация данных

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

Используя FastAPI Pagination, мы можем предоставить пользователям удобный интерфейс для работы с большими объемами данных, улучшить производительность и снизить нагрузку на сервер.

FastAPI Cache2: Кэширование данных

FastAPI Cache2 — это библиотека для кэширования данных в FastAPI. Кэширование позволяет значительно сократить время ответа на повторяющиеся запросы, снижая нагрузку на базу данных и улучшая производительность приложения.

В нашем проекте FastAPI Cache2 используется для кэширования результатов частых запросов, что позволяет ускорить работу приложения и снизить издержки на вычислительные ресурсы.

Docker: Контейнеризация приложения

Docker — это инструмент для контейнеризации, который позволяет изолировать приложение и все его зависимости в отдельной среде. Это упрощает процесс развертывания и обеспечивает совместимость кода на разных системах.

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

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

1. Создание виртуального окружения:

   python3 -m venv venv

   Эта команда создаст виртуальное окружение в директории venv. После его создания необходимо активировать окружение:

   • Для Linux/MacOS:

     source venv/bin/activate

   • Для Windows:

     venv\Scripts\activate

2. Установка зависимостей: После активации виртуального окружения можно установить все необходимые зависимости. Для этого создается файл requirements.txt, который будет содержать все используемые библиотеки:

   pip install fastapi[all] fastapi-users[sqlalchemy] fastapi-pagination fastapi-cache2 sqladmin[full] gunicorn aioboto3 pillow filetype sqlalchemy[asyncio] sqlalchemy-utils alembic asyncpg psycopg2-binary redis[asyncio] taskiq-fastapi taskiq-aio-pika taskiq-redis celery celery-sqlalchemy-scheduler pytest pytest-asyncio

   После установки всех зависимостей можно зафиксировать их версии в файле requirements.txt:

   pip freeze > requirements.txt

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

После настройки окружения важно правильно организовать структуру проекта. Четко продуманная структура не только упрощает разработку, но и способствует поддерживаемости и масштабируемости приложения в будущем. Рассмотрим типичную структуру FastAPI проекта:

backend/
└── alembic/                # Директория для миграций базы данных
└── src/                    # Основная директория с исходным кодом
│   ├── admin/              # Модули для административных функций
│   ├── auth/               # Модули для аутентификации и авторизации
│   ├── conf/               # Конфигурационные файлы и классы
│   ├── crud/               # Реализация CRUD операций
│   ├── db/                 # Модули для работы с базой данных
│   ├── media/              # Работа с медиафайлами (загрузка и хранение)
│   ├── models/             # Определение моделей базы данных
│   ├── schemas/            # Определение схем данных для API
│   ├── static/             # Статические файлы (CSS, JS, изображения)
│   ├── tasks/              # Модули для фоновых задач
│   ├── templates/          # HTML-шаблоны для рендеринга (если используется)
│   ├── users/              # Работа с пользователями
│   ├── __init__.py         # Файл инициализации пакета
│   ├── cli.py              # Скрипты для командной строки (управление приложением)
│   ├── alembic.ini         # Конфигурация Alembic
│   ├── main.py             # Главный файл приложения (точка входа)
│   ├── .dockerignore       # Настройка исключения файлов при копировании в докер контейнер 
│   ├── routers.py          # Интеграция маршрутов из других частей приложения в единый router
│   ├── Dockerfile          # Файл для создания Docker-образа
│   └── requirements.txt    # Файл зависимостей
infra/
│   ├── stage/   
│ 	│   └── docker-compose.yml  # Конфигурация для Docker Compose
venv/                           # Виртуальное окружение (не добавляется в репозиторий)
.env                            # Файл конфигурации окружения
.gitignore                      # Файл игнорирования файлов для Git
LICENSE.md                      # Лицензия проекта
README.md                       # Описание проекта

Каждая директория и файл в этой структуре имеет свое назначение:

• alembic/ — директория для управления миграциями базы данных, используя Alembic.
• src/ — основная директория с исходным кодом приложения, где каждая подсистема организована в отдельный модуль.
• conf/ — директория для конфигурационных файлов и классов, что позволяет централизованно управлять настройками проекта.
• crud/ — содержит реализацию операций CRUD, что обеспечивает удобное и безопасное взаимодействие с базой данных.
• models/ и schemas/ — директории для определения моделей базы данных и схем данных, что способствует четкому разделению логики и данных.
• tasks/ — модуль для работы с фоновыми задачами, что позволяет разгрузить основное приложение и обрабатывать тяжелые операции асинхронно.

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

Файл .env служит для хранения конфиденциальных данных и конфигурационных параметров, которые могут меняться в зависимости от среды (например, development, production). Хранение этих данных в отдельном файле позволяет изолировать настройки, которые могут быть специфичны для каждой среды, и при этом не включать их в систему контроля версий (например, Git).

# project folder
PYTHONPATH=backend

#redis
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=redis_password

# rabbitmq
RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_DEFAULT_USER=user_rabbit
RABBITMQ_DEFAULT_PASS=rabbit_password

# postgres
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DB=db_name
POSTGRES_USER=db_user
POSTGRES_PASSWORD=postgres_password

# minio
MINIO_STATIC_BUCKET=static
MINIO_MEDIA_BUCKET=media
MINIO_DATABASE_BUCKET=backup

MINIO_ROOT_USER=myrootaccesskey
MINIO_ROOT_PASSWORD=myrootpassword

MINIO_ACCESS_KEY=myrootaccesskey
MINIO_SECRET_KEY=myrootpassword

MINIO_DOMAIN=aws.localhost
MINIO_USE_SSL=0
MINIO_REGION_NAME=ru-central1

# FastApi settings
MODE=development

SECRET_KEY='fastapi-insecure-secret_key'

BACKEND_CORS_ORIGINS=http://localhost,http://127.0.0.1

Этот файл содержит все необходимые параметры для настройки подключения к Redis, RabbitMQ и PostgreSQL, а также другие важные настройки, такие как секретный ключ и разрешенные источники для CORS. В случае использования различных сред (development, testing, production), можно создавать несколько .env файлов, каждый из которых будет содержать свои специфические настройки.

Для управления конфигурацией в FastAPI проекте используется класс Settings, который наследуется от BaseSettings из Pydantic. Этот класс позволяет загружать параметры из файла .env, валидировать их и предоставлять удобный интерфейс для работы с настройками.

import base64
import os
import secrets
from enum import Enum
from pathlib import Path
from typing import Any

from dotenv import load_dotenv
from pydantic import AnyHttpUrl, PostgresDsn, field_validator
from pydantic_core.core_schema import FieldValidationInfo
from pydantic_settings import BaseSettings, SettingsConfigDict

load_dotenv()


class ModeEnum(str, Enum):
    development = "development"
    production = "production"
    testing = "testing"


class Settings(BaseSettings):
    MODE: str = os.environ.get("MODE")

    REDIS_HOST: str = os.environ.get("REDIS_HOST")
    REDIS_PORT: str = os.environ.get("REDIS_PORT")
    REDIS_PASSWORD: str = os.environ.get("REDIS_PASSWORD")
    DB_POOL_SIZE: int = 83
    WEB_CONCURRENCY: int = 9
    POOL_SIZE: int = max(DB_POOL_SIZE // WEB_CONCURRENCY, 5)

    RABBITMQ_HOST: str = os.getenv('RABBITMQ_HOST')
    RABBITMQ_PORT: str = os.getenv('RABBITMQ_PORT')
    RABBITMQ_DEFAULT_USER: str = os.getenv('RABBITMQ_DEFAULT_USER')
    RABBITMQ_DEFAULT_PASS: str = os.getenv('RABBITMQ_DEFAULT_PASS')

    POSTGRES_HOST: str = os.getenv("POSTGRES_HOST")
    POSTGRES_PORT: int = os.getenv("POSTGRES_PORT")
    POSTGRES_DB: str = os.getenv("POSTGRES_DB")
    POSTGRES_USER: str = os.getenv("POSTGRES_USER")
    POSTGRES_PASSWORD: str = os.getenv("POSTGRES_PASSWORD")
    DATABASE_CELERY_NAME: str = "celery_schedule_jobs"

    MINIO_ACCESS_KEY: str = os.getenv("MINIO_ACCESS_KEY")
    MINIO_SECRET_KEY: str = os.getenv("MINIO_SECRET_KEY")
    MINIO_DOMAIN: str = os.getenv("MINIO_DOMAIN")
    MINIO_REGION_NAME: str = os.getenv("MINIO_REGION_NAME")
    MINIO_MEDIA_BUCKET: str = os.getenv("MINIO_MEDIA_BUCKET")
    MINIO_STATIC_BUCKET: str = os.getenv("MINIO_STATIC_BUCKET")
    MINIO_DATABASE_BUCKET: str = os.getenv("MINIO_DATABASE_BUCKET")
    MINIO_USE_SSL: bool = int(os.getenv("MINIO_USE_SSL"))

    SECRET_KEY: str = os.getenv("SECRET_KEY", secrets.token_urlsafe(32))
    ENCRYPT_KEY: str = os.getenv("ENCRYPT_KEY", base64.urlsafe_b64encode(os.urandom(32)).decode()) # если используется

    CELERY_BROKER_URL: str = ""
    CELERY_RESULT_BACKEND: str = ""

    @field_validator("CELERY_BROKER_URL", "CELERY_RESULT_BACKEND", mode="after")
    def assemble_celery_urls(cls, v: str | None, info: FieldValidationInfo):
        if all(info.data.get(attr) for attr in ["RABBITMQ_DEFAULT_USER", "RABBITMQ_DEFAULT_PASS", "RABBITMQ_HOST", "RABBITMQ_PORT"]):
            url = (
                f"amqp://{info.data['RABBITMQ_DEFAULT_USER']}:"
                f"{info.data['RABBITMQ_DEFAULT_PASS']}@"
                f"{info.data['RABBITMQ_HOST']}:"
                f"{info.data['RABBITMQ_PORT']}//"
            )
            if info.field_name == "CELERY_BROKER_URL":
                return url
            elif info.field_name == "CELERY_RESULT_BACKEND":
                return url
        return v

    ASYNC_DATABASE_URI: PostgresDsn | str = ""

    @field_validator("ASYNC_DATABASE_URI", mode="after")
    def assemble_async_db_connection(cls, v: str | None, info: FieldValidationInfo) -> Any:
        if isinstance(v, str):
            if v == "":
                return PostgresDsn.build(
                    scheme="postgresql+asyncpg",
                    username=info.data["POSTGRES_USER"],
                    password=info.data["POSTGRES_PASSWORD"],
                    host=info.data["POSTGRES_HOST"],
                    port=info.data["POSTGRES_PORT"],
                    path=info.data["POSTGRES_DB"],
                )
        return v

    SYNC_DATABASE_URI: PostgresDsn | str = ""

    @field_validator("SYNC_DATABASE_URI", mode="after")
    def assemble_sync_db_connection(cls, v: str | None, info: FieldValidationInfo) -> Any:
        if isinstance(v, str):
            if v == "":
                return PostgresDsn.build(
                    scheme="postgresql+psycopg2",
                    username=info.data["POSTGRES_USER"],
                    password=info.data["POSTGRES_PASSWORD"],
                    host=info.data["POSTGRES_HOST"],
                    port=info.data["POSTGRES_PORT"],
                    path=info.data["POSTGRES_DB"],
                )
        return v

    SYNC_CELERY_DATABASE_URI: PostgresDsn | str = ""

    @field_validator("SYNC_CELERY_DATABASE_URI", mode="after")
    def assemble_celery_db_connection(
        cls, v: str | None, info: FieldValidationInfo
    ) -> Any:
        if isinstance(v, str):
            if v == "":
                return PostgresDsn.build(
                    scheme="db+postgresql",
                    username=info.data["POSTGRES_USER"],
                    password=info.data["POSTGRES_PASSWORD"],
                    host=info.data["POSTGRES_HOST"],
                    port=info.data["POSTGRES_PORT"],
                    path=info.data["DATABASE_CELERY_NAME"],
                )
        return v

    SYNC_CELERY_BEAT_DATABASE_URI: PostgresDsn | str = ""

    @field_validator("SYNC_CELERY_BEAT_DATABASE_URI", mode="after")
    def assemble_celery_beat_db_connection(
        cls, v: str | None, info: FieldValidationInfo
    ) -> Any:
        if isinstance(v, str):
            if v == "":
                return PostgresDsn.build(
                    scheme="postgresql+psycopg2",
                    username=info.data["POSTGRES_USER"],
                    password=info.data["POSTGRES_PASSWORD"],
                    host=info.data["POSTGRES_HOST"],
                    port=info.data["POSTGRES_PORT"],
                    path=info.data["DATABASE_CELERY_NAME"],
                )
        return v

    BACKEND_CORS_ORIGINS: str | list[AnyHttpUrl] = os.getenv("BACKEND_CORS_ORIGINS", "http://localhost")

    @field_validator("BACKEND_CORS_ORIGINS", mode="after")
    def assemble_cors_origins(cls, v: str | list[str]) -> list[str] | str:
        if isinstance(v, str) and not v.startswith("["):
            return [i.strip() for i in v.split(",")]
        elif isinstance(v, (list, str)):
            return v
        raise ValueError(v)

    project_root: Path = Path(__file__).parent.parent.resolve()
    model_config = SettingsConfigDict(
        case_sensitive=True, env_file=os.path.expanduser("~/.env")
    )


settings = Settings()

Класс Settings загружает переменные окружения из файла .env и предоставляет доступ к ним через атрибуты класса. Такой подход позволяет централизованно управлять всеми конфигурационными параметрами приложения. Более того, использование Pydantic для валидации данных обеспечивает корректность настроек и предупреждает возможные ошибки на этапе их загрузки.

Одной из ключевых возможностей, предоставляемых классом Settings, является валидация и обработка конфигурационных данных. Использование Pydantic позволяет автоматически проверять значения параметров на соответствие ожидаемым типам, а также осуществлять более сложную логику валидации с помощью валидаторов (field_validator).

@field_validator("ASYNC_DATABASE_URI", mode="after")
def assemble_async_db_connection(cls, v: str | None, info: FieldValidationInfo) -> Any:
    if isinstance(v, str) and v == "":
        return PostgresDsn.build(
            scheme="postgresql+asyncpg",
            username=info.data["POSTGRES_USER"],
            password=info.data["POSTGRES_PASSWORD"],
            host=info.data["POSTGRES_HOST"],
            port=info.data["POSTGRES_PORT"],
            path=info.data["POSTGRES_DB"],
        )
    return v

Этот валидатор проверяет, была ли строка подключения к базе данных пустой, и если да, то собирает ее, используя данные, указанные в других параметрах (например, POSTGRES_USER, POSTGRES_PASSWORD и т.д.). Такой подход позволяет не дублировать информацию в конфигурации и минимизировать количество потенциальных ошибок.

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

Кроме того, использование файлов .env для хранения конфиденциальных данных и параметров окружения делает проект более безопасным и гибким. Это позволяет легко адаптировать приложение к различным средам (development, production, testing), просто изменяя значения в файле .env.

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

Dockerfile — это сценарий, содержащий инструкции для сборки Docker-образа вашего приложения. Рассмотрим пошагово пример Dockerfile для FastAPI проекта:

FROM python:3.11-alpine

WORKDIR /app

RUN apk add --no-cache \
    postgresql-client

COPY requirements.txt .

RUN pip install -r requirements.txt --no-cache-dir

COPY ./src /app/src
COPY ./alembic /app/alembic
COPY alembic.ini .

RUN addgroup -S app-group && \
    adduser -S -G app-group app-user && \
    chown -R app-user:app-group /app

USER app-user

ENV PYTHONPATH=/app/src

Разбор Dockerfile:

1. Базовый образ:

   FROM python:3.11-alpine

   Мы используем легковесный образ python:3.11-alpine в качестве базового. Alpine — это минималистичный дистрибутив Linux, который значительно снижает размер итогового Docker-образа.

2. Рабочая директория:

   WORKDIR /app

   Устанавливаем рабочую директорию /app, в которую будут копироваться файлы проекта и где будет выполняться приложение.

3. Установка зависимостей:

   RUN apk add --no-cache \
       postgresql-client

   Используем apk, пакетный менеджер Alpine, для установки postgresql-client, который необходим для работы с базой данных в контейнере (метод backup и restore db).

4. Установка Python-зависимостей:

   Сначала копируем файл зависимостей requirements.txt в контейнер, затем устанавливаем указанные в нем зависимости. Флаг --no-cache-dir предотвращает сохранение временных файлов, что уменьшает размер Docker-образа.

   COPY requirements.txt .
   RUN pip install -r requirements.txt --no-cache-dir

5. Копирование файлов проекта:

   COPY ./src /app/src
   COPY ./alembic /app/alembic
   COPY alembic.ini .

   Копируем исходный код приложения, директорию с миграциями Alembic и файл конфигурации Alembic в контейнер.

6. Создание пользователя и установка прав:

   RUN addgroup -S app-group && \
       adduser -S -G app-group app-user && \
       chown -R app-user:app-group /app
   USER app-user

   Для повышения безопасности создаем группу и пользователя внутри контейнера, затем устанавливаем права на директорию /app. После этого переключаемся на созданного пользователя.

7. Настройка переменных окружения:

   ENV PYTHONPATH=/app/src

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

docker-compose.yml позволяет одновременно управлять несколькими контейнерами, определяя, как они взаимодействуют друг с другом. Этот файл особенно полезен для разработки и тестирования, так как позволяет запускать весь стек технологий командой docker-compose up.

x-shared-parameters: &shared-parameters
  env_file:
    - ../../.env
  network_mode: "host"
  restart: always

x-db-dependency: &db-dependency
  depends_on:
    db:
      condition: service_started
    rabbitmq:
      condition: service_healthy

services:
  db:
    <<: *shared-parameters
    image: postgres:16-alpine
    container_name: tmpl_stage_db
    volumes:
      - tmpl_stage_psgsql_volume:/var/lib/postgresql/data/
    ports:
      - "5432:5432"

  pgadmin:
    <<: *shared-parameters
    image: dpage/pgadmin4:latest
    container_name: pgadmin
    ports:
      - "8090:8090"
    volumes:
      - pgadmin_volume:/var/lib/pgadmin/
    depends_on:
      - db

  redis:
    <<: *shared-parameters
    image: redis:latest
    container_name: tmpl_stage_redis
    command: >
      --requirepass ${REDIS_PASSWORD}
    ports:
      - "6379:6379"

  rabbitmq:
    <<: *shared-parameters
    image: rabbitmq:management-alpine
    container_name: tmpl_stage_rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"
    volumes:
      - tmpl_stage_rabbitmq_volume:/var/lib/rabbitmq/:rw
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5

  backend:
    <<: [*shared-parameters, *db-dependency]
    build:
      context: ../../backend
      dockerfile: Dockerfile
    container_name: tmpl_stage_backend
    command: ./src/conf/entrypoints/fastapi.sh
    ports:
      - 8100:8100

  celery:
    <<: [*shared-parameters, *db-dependency]
    build:
      context: ../../backend
      dockerfile: Dockerfile
    container_name: tmpl_stage_celery
    command: ./src/conf/entrypoints/celery.sh

  celery-beat:
    <<: [*shared-parameters, *db-dependency]
    build:
      context: ../../backend
      dockerfile: Dockerfile
    container_name: tmpl_stage_celery-beat
    command: ./src/conf/entrypoints/celery-beat.sh
    mem_limit: 1g
    cpus: "1.0"

  taskiq:
    <<: [*shared-parameters, *db-dependency]
    build:
      context: ../../backend
      dockerfile: Dockerfile
    container_name: tmpl_stage_taskiq_worker
    command: ./src/conf/entrypoints/taskiq.sh

volumes:
  tmpl_stage_psgsql_volume:
  pgadmin_volume:
  tmpl_stage_rabbitmq_volume:

Разбор docker-compose.yml:

1. Общие параметры и переменные окружения:

   x-shared-parameters: &shared-parameters
     env_file:
       - ../../.env
     network_mode: "host"
     restart: always

   Использование YAML-анкоров (&shared-parameters) и ссылок (<<: *shared-parameters) позволяет определить общие параметры для всех сервисов, такие как подключение к файлу переменных окружения .env, режим работы сети и параметры перезапуска.  network_mode: "host" В нашем случае используется исключительно в разработке. В продакшен среде, сеть лучше создавать внутри docker.

2. База данных PostgreSQL:

   db:
     <<: *shared-parameters
     image: postgres:16-alpine
     container_name: tmpl_stage_db
     volumes:
       - tmpl_stage_psgsql_volume:/var/lib/postgresql/data/
     ports:
       - "5432:5432"

   Контейнер с PostgreSQL на основе легковесного образа postgres:16-alpine. Данные базы данных сохраняются на хост-машине в volume tmpl_stage_psgsql_volume, что обеспечивает их сохранение между запусками контейнеров.

3. Redis:

   redis:
     <<: *shared-parameters
     image: redis:latest
     container_name: tmpl_stage_redis
     command: >
       --requirepass ${REDIS_PASSWORD}
     ports:
       - "6379:6379"

   Контейнер с Redis, который будет использоваться для кэширования данных и работы с очередями сообщений. Для повышения безопасности, доступ к Redis защищен паролем.

4. RabbitMQ:

   rabbitmq:
     <<: *shared-parameters
     image: rabbitmq:management-alpine
     container_name: tmpl_stage_rabbitmq
     ports:
       - "5672:5672"
       - "15672:15672"
     volumes:
       - tmpl_stage_rabbitmq_volume:/var/lib/rabbitmq/:rw
     healthcheck:
       test: ["CMD", "rabbitmq-diagnostics", "ping"]
       interval: 10s
       timeout: 5s
       retries: 5

   Контейнер с RabbitMQ, который управляет очередями сообщений. Включена проверка состояния контейнера с помощью healthcheck, что позволяет убедиться, что сервис работает корректно перед запуском зависимых от него контейнеров.

5. Backend (FastAPI):

   backend:
     <<: [*shared-parameters, *db-dependency]
     build:
       context: ../../backend
       dockerfile: Dockerfile
     container_name: tmpl_stage_backend
     command: ./src/conf/entrypoints/fastapi.sh
     ports:
       - 8100:8100

   Основной контейнер, который запускает FastAPI приложение. Важным моментом является использование скрипта (fastapi.sh) для инициализации приложения, что позволяет гибко управлять процессом его запуска из проекта, не изменяя docker-compose.yml.

6. Celery и TaskIQ: Контейнеры celery, celery-beat и taskiq управляют фоновыми задачами. Они также используют общее окружение и зависимости от базы данных и RabbitMQ. Ограничение памяти и процессорных ресурсов для контейнера celery-beat помогает контролировать нагрузку на систему.

7. Volumes:

   volumes:
     tmpl_stage_psgsql_volume:
     pgadmin_volume:
     tmpl_stage_rabbitmq_volume:

   Использование volumes для хранения данных позволяет контейнерам сохранять свое состояние между перезапусками и обеспечивает безопасность данных.

Параметры сборки и запуска приложения

Для удобства разработки и развертывания проекта используется несколько стратегий:

1. Изоляция окружений:
   • Использование .env файла позволяет настраивать контейнеры в зависимости от окружения (development, production). Это облегчает переключение между различными конфигурациями и поддерживает безопасность, скрывая чувствительные данные.
2. Удобство разработки:
   • Функция network_mode: "host" в docker-compose.yml обеспечивает прямой доступ к хост-сетям, что упрощает взаимодействие между контейнерами и хост-системой. Это особенно полезно в процессе разработки и тестирования.
3. Проверка готовности сервисов:
   • Использование depends_on и healthcheck позволяет убедиться, что зависимые сервисы (например, база данных и брокер сообщений) полностью готовы перед запуском основных контейнеров. Это улучшает стабильность системы и предотвращает ошибки запуска.

Модели данных лежат в основе любого приложения, взаимодействующего с базой данных. В FastAPI проекте, используя SQLAlchemy, мы можем определять структуры таблиц, устанавливать связи между ними и управлять данными с высокой степенью контроля и гибкости. В этом разделе мы рассмотрим подход к проектированию моделей данных, особенности работы с асинхронными базами данных и преимущества использования SQLAlchemy в современных приложениях.

SQLAlchemy — это ORM (Object-Relational Mapping) инструмент, который предоставляет разработчикам Python удобные механизмы для работы с базами данных. Основное преимущество SQLAlchemy заключается в том, что он позволяет описывать структуру базы данных и её отношения с помощью Python-классов, которые затем автоматически транслируются в SQL-запросы.

Каждая модель в SQLAlchemy представляет собой класс, который наследуется от базового класса. Этот класс описывает структуру таблицы в базе данных, включая поля, их типы и ограничения. В нашем проекте все модели наследуются от базового класса Base, который обеспечивает общие поля и поведение для всех моделей:

from datetime import datetime
from uuid import UUID, uuid4

from sqlalchemy import DateTime, func
from sqlalchemy.dialects.postgresql import UUID as UUIDType
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    id: Mapped[UUID] = mapped_column(UUIDType(as_uuid=True), primary_key=True, default=uuid4, index=True, nullable=False)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=func.now(), onupdate=func.now())
    created_at: Mapped[datetime] = mapped_column(DateTime, default=func.now())

    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

Разбор базовой модели:

1. Поле id:
   • id — это уникальный идентификатор каждой записи, который представлен в виде UUID. UUID выбирается вместо автоинкрементного целочисленного идентификатора для повышения безопасности и уменьшения риска коллизий при масштабировании приложения.
2. Поля created_at и updated_at:
   • Эти поля автоматически заполняются при создании и обновлении записи с помощью функций func.now() SQLAlchemy. Они полезны для отслеживания времени создания и последнего обновления записей.
3. Метод __tablename__:
   • Метод __tablename__ автоматически устанавливает имя таблицы в базе данных в нижнем регистре, используя имя класса. Это обеспечивает единообразие и упрощает управление схемой базы данных.

Модель User — это пример более сложной структуры данных, которая включает как простые текстовые поля, так и связи с другими таблицами. В нашем проекте для работы с пользователями используется библиотека fastapi-users, которая облегчает управление пользователями, включая регистрацию, авторизацию и управление правами доступа.

from fastapi_users.db import SQLAlchemyBaseUserTable
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.models.base_model import Base
from src.models.interim_tables import user_image_association


class User(SQLAlchemyBaseUserTable, Base):
    first_name: Mapped[str] = mapped_column(String, nullable=True)
    last_name: Mapped[str] = mapped_column(String, nullable=True)

    image_files = relationship("Image", secondary=user_image_association, back_populates="users")

    def __repr__(self):
        return f"<User(id={self.id}, first_name={self.first_name}, last_name={self.last_name})>"

Разбор модели User:

1. Наследование от SQLAlchemyBaseUserTable:
   • Модель User наследуется от SQLAlchemyBaseUserTable, предоставляемого fastapi-users. Это позволяет использовать встроенные функции для управления пользователями, такие как хеширование паролей и проверка данных.
2. Поля first_name и last_name:
   • Эти поля хранят имена пользователей и могут быть пустыми (nullable). Они добавляются к основным полям, предоставляемым базовой таблицей пользователей.
3. Связи через relationship:
   • Поле image_files представляет связь многие-ко-многим между пользователями и изображениями. Используя таблицу ассоциаций user_image_association, эта связь управляется автоматически, что упрощает работу с множественными связанными записями. Но об этом в следующей статье.
4. Метод __repr__:
   • Переопределение метода __repr__ улучшает читаемость логов и отладочной информации, предоставляя краткое представление о пользователе через его идентификатор и имена.

В современных веб-приложениях важную роль играет асинхронная работа с базой данных, которая позволяет обрабатывать множество запросов одновременно, не блокируя основной поток выполнения. Это особенно актуально для приложений, работающих под высокой нагрузкой.

В FastAPI проекте для работы с базой данных используется асинхронная версия SQLAlchemy. Это требует настройки соответствующих движков и сессий:

from sqlalchemy import QueuePool, create_engine
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio.session import AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import AsyncAdaptedQueuePool, NullPool
from src.conf import ModeEnum, settings

DB_POOL_SIZE = 83
WEB_CONCURRENCY = 9
POOL_SIZE = max(DB_POOL_SIZE // WEB_CONCURRENCY, 5)


engine = create_async_engine(
    str(settings.ASYNC_DATABASE_URI),
    echo=False,
    poolclass=NullPool if settings.MODE == ModeEnum.testing else AsyncAdaptedQueuePool,
    pool_size=POOL_SIZE,
    max_overflow=64,
)
async_session = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Разбор асинхронной конфигурации:

1. Асинхронный движок create_async_engine:
   • Движок создается с помощью функции create_async_engine, которая позволяет управлять асинхронными соединениями к базе данных. В качестве параметров передаются строка подключения и параметры пула соединений.
2. Асинхронная сессия async_session:
   • Сессия создается с использованием sessionmaker и класса AsyncSession. Это позволяет безопасно управлять транзакциями и асинхронно выполнять запросы к базе данных.
3. Пул соединений:
   • Использование пулов соединений (AsyncAdaptedQueuePool или NullPool) позволяет управлять доступом к базе данных в многопоточных приложениях, что улучшает производительность и надежность.

Использование SQLAlchemy в связке с асинхронными базами данных предоставляет разработчикам множество преимуществ:

1. Высокая производительность:
   • Асинхронные операции позволяют обрабатывать множество запросов одновременно, не блокируя основную логику приложения. Это особенно важно для высоконагруженных систем.
2. Удобство и гибкость:
   • SQLAlchemy предоставляет гибкий интерфейс для работы с базами данных, позволяя разработчикам сосредоточиться на логике приложения, а не на низкоуровневых деталях работы с SQL.
3. Расширяемость:
   • Благодаря ORM, модели данных могут быть легко расширены и адаптированы под новые требования. Это упрощает добавление новых функций и поддержание кода.
4. Управление связями и транзакциями:
   • SQLAlchemy автоматизирует управление сложными связями между таблицами и транзакциями, что уменьшает количество потенциальных ошибок и улучшает целостность данных.

Веб-приложения зачастую основываются на взаимодействии с базой данных, где необходимо выполнять основные операции: создание, чтение, обновление и удаление (CRUD). Для упрощения и унификации работы с различными моделями данных в FastAPI проекте был реализован обобщённый класс GenericCRUD, который предоставляет стандартные CRUD-методы. В этом разделе мы рассмотрим подход к реализации CRUD-операций с использованием этого класса и обсудим, как типизация в Python способствует повышению безопасности и удобства работы с кодом.

Обобщённый класс GenericCRUD — это шаблон, который можно использовать для работы с любыми моделями данных, наследующимися от базового класса Base. Этот класс предоставляет базовые методы для выполнения CRUD-операций, минимизируя количество повторяющегося кода и повышая его переиспользуемость. Отдельное спасибо Jonathan Vargas за идею!

from typing import Any, Generic, Type, TypeVar
from uuid import UUID

from fastapi import HTTPException
from fastapi_pagination import Page, Params
from fastapi_pagination.ext.sqlalchemy import paginate
from pydantic import BaseModel
from sqlalchemy import exc, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.selectable import Select
from src.conf import logger
from src.models.base_model import Base

ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
logger = logger.getChild(__name__)

Разбор кода:

1. Параметризация класса:
   • Класс GenericCRUD параметризован с использованием TypeVar для указания типов моделей данных (ModelType), схем создания (CreateSchemaType) и схем обновления (UpdateSchemaType). Это позволяет использовать один и тот же класс для работы с различными моделями, сохраняя при этом строгую типизацию и безопасность.
2. Типизация:
   • Типизация играет ключевую роль в обеспечении корректности кода. Использование типовых переменных позволяет IDE и инструментам статического анализа кода (таким как mypy) проверять соответствие типов и предотвращать ошибки на этапе компиляции, что значительно повышает надежность приложения.

Рассмотрим методы класса GenericCRUD, отвечающие за выполнение основных CRUD-операций:

Метод get: получение записи по идентификатору

async def get(self, *, id: UUID | str, db_session: AsyncSession | None = None) -> ModelType | None:
    query = select(self.model).where(self.model.id == id)
    result = await db_session.execute(query)
    return result.scalars().one_or_none()

Этот метод выполняет запрос к базе данных для получения записи по её идентификатору. Если запись существует, она возвращается, иначе метод возвращает None.

Метод create: создание новой записи

async def create(
    self, *, obj_in: CreateSchemaType | ModelType, created_by_id: UUID | str | None = None, db_session: AsyncSession | None = None,
) -> ModelType:

    if not isinstance(obj_in, self.model):
        db_obj = self.model(**obj_in.model_dump())
    else:
        db_obj = obj_in

    if created_by_id and hasattr(db_obj, 'created_by_id'):
        db_obj.id = created_by_id

    try:
        db_session.add(db_obj)
        await db_session.commit()
    except exc.IntegrityError:
        db_session.rollback()
        raise HTTPException(status_code=409, detail="Resource already exists")
    await db_session.refresh(db_obj)
    return db_obj

Этот метод создает новую запись в базе данных. Перед добавлением в базу данных проверяется, является ли переданный объект экземпляром целевой модели. Если объект уже существует (например, если возникла ошибка целостности данных), операция откатывается, и возвращается HTTP-ошибка.

Метод update: обновление существующей записи

async def update(
    self, *, obj_current: ModelType, obj_new: UpdateSchemaType | dict[str, Any] | ModelType, db_session: AsyncSession | None = None,
) -> ModelType:

    if isinstance(obj_new, dict):
        update_data = obj_new
    else:
        update_data = obj_new.model_dump(exclude_unset=True)
    for field in update_data:
        setattr(obj_current, field, update_data[field])

    db_session.add(obj_current)
    await db_session.commit()
    await db_session.refresh(obj_current)
    return obj_current

Метод update позволяет обновлять существующую запись в базе данных. Новые данные могут быть представлены как в виде объекта, так и в виде словаря. Этот метод обновляет только те поля, которые были явно изменены, что делает его гибким и эффективным.

Метод remove: удаление записи по идентификатору

async def remove(
    self, *, id: UUID | str, db_session: AsyncSession | None = None
) -> ModelType:
    result = await db_session.execute(select(self.model).where(self.model.id == id))
    obj = result.scalars().one()
    await db_session.delete(obj)
    await db_session.commit()
    return obj

Метод remove выполняет удаление записи из базы данных по её идентификатору. Перед удалением запись извлекается из базы, чтобы убедиться, что она существует.

Для удобства работы с большими объемами данных в GenericCRUD реализован метод get_multi_paginated, который поддерживает пагинацию с использованием библиотеки fastapi_pagination.

async def get_multi_paginated(
    self, *, params: Params | None = Params(), query: T | Select[T] | None = None, db_session: AsyncSession | None = None,
) -> Page[ModelType]:
    if query is None:
        query = select(self.model)

    output = await paginate(db_session, query, params)
    return output

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

Типизация в Python — это не просто удобство, но и важный инструмент для обеспечения безопасности кода. В контексте GenericCRUD типизация позволяет:

1. Снизить количество ошибок:
   • IDE и линтеры могут автоматически выявлять несоответствия типов, что снижает вероятность ошибок в коде.
2. Улучшить читаемость и поддержку кода:
   • Явная типизация делает код более понятным и поддерживаемым, так как разработчики сразу видят, с какими типами данных работают методы.
3. Повысить надежность и предсказуемость:
   • Типизированные методы и классы позволяют разработчикам быть уверенными, что данные, передаваемые между различными частями приложения, соответствуют ожидаемым структурам и типам.

Кэширование — это эффективный способ оптимизации веб-приложений, который позволяет значительно снизить нагрузку на сервер, уменьшив количество повторяющихся запросов к базе данных или внешним API. В FastAPI кэширование можно легко реализовать с помощью библиотеки fastapi-cache2 и Redis.

Преимущества кэширования:

1. Уменьшение времени отклика:
   • Данные, сохраненные в кэше, могут быть извлечены значительно быстрее, чем из базы данных. Это особенно важно для данных, которые редко изменяются, но часто запрашиваются.
2. Снижение нагрузки на сервер:
   • За счет кэширования уменьшается количество запросов к базе данных, что позволяет уменьшить нагрузку на сервер и повысить его производительность.
3. Повышение устойчивости приложения:
   • В случае временной недоступности базы данных или внешнего API, кэшированные данные могут использоваться для продолжения обслуживания запросов, что повышает устойчивость системы.

В проекте кэширование реализуется с использованием Redis в качестве хранилища кэша:

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from src.conf.redis import set_async_redis_client

@asynccontextmanager
async def lifespan(app: FastAPI):
    redis_client = await set_async_redis_client()
    FastAPICache.init(RedisBackend(redis_client), prefix="fastapi-cache")
    yield

app = FastAPI(lifespan=lifespan)

В этом коде инициализация кэша происходит в рамках lifespan приложения. Используя RedisBackend, данные сохраняются в Redis, а FastAPICache обеспечивает интерфейс для работы с кэшем.

Пример использования кэша в методах CRUD может выглядеть следующим образом:

from fastapi_cache.decorator import cache

@cache(expire=60)
async def get_user_by_id(user_id: UUID, db_session: AsyncSession):
    user = await crud_user.get(id=user_id, db_session=db_session)
    return user

Здесь метод get_user_by_id будет использовать кэш для хранения результатов на 60 секунд. Если пользователь с данным user_id был запрошен ранее, результат будет извлечен из кэша, минуя запрос к базе данных.

Для работы с Redis в асинхронном контексте используется модуль redis.asyncio. Это позволяет взаимодействовать с Redis без блокировки основного потока выполнения, что особенно важно в асинхронных приложениях.

import redis.asyncio as aioredis
from redis.asyncio import Redis

class AsyncRedisClient:
    _client: Redis = None

    @classmethod
    async def initialize(cls):
        from . import settings
        if cls._client is None:
            cls._client = await aioredis.from_url(
                f"redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:{settings.REDIS_PORT}",
                max_connections=20,
                encoding="utf8",
                decode_responses=True,
                socket_connect_timeout=5,
                socket_timeout=5,
            )
        return cls._client

    @classmethod
    async def get_client(cls):
        if cls._client is None:
            await cls.initialize()
        return cls._client


async def set_async_redis_client() -> Redis:
    client = await AsyncRedisClient.initialize()
    logger.info("AsyncRedisClient is setting...")
    return client

Этот класс реализует подключение к Redis и управление асинхронным клиентом. Он используется для инициализации и получения клиента Redis, который затем может быть использован для выполнения кэширования или других операций. В данном примере кода используется паттерн Singleton (Одиночка). Singleton — это порождающий паттерн, который гарантирует, что у класса будет только один экземпляр, и предоставляет глобальную точку доступа к этому экземпляру.

fastapi-users — это библиотека, специально разработанная для интеграции системы управления пользователями в приложения на FastAPI. Она предоставляет готовые решения для регистрации, аутентификации, восстановления паролей и верификации пользователей, что значительно упрощает процесс разработки.

Основные возможности библиотеки включают:

• Поддержка различных стратегий аутентификации (JWT, OAuth, Bearer Token и др.).
• Готовые маршруты для регистрации, входа, сброса пароля и верификации пользователей.
• Поддержка асинхронных баз данных, что идеально подходит для современных высоконагруженных приложений.
• Гибкость в настройке и расширении функциональности через пользовательские менеджеры и адаптеры баз данных.

В нашем проекте система управления пользователями реализована с использованием fastapi-users и асинхронной базы данных на основе SQLAlchemy. Давайте рассмотрим ключевые элементы этой реализации.

Менеджер пользователей отвечает за управление учетными записями пользователей, включая регистрацию, обновление данных, загрузку изображений и другие операции. В нашем проекте UserManager реализован как класс, наследующийся от BaseUserManager и UUIDIDMixin, что позволяет использовать UUID в качестве идентификаторов пользователей.

import uuid
from typing import Optional

from fastapi import Depends, Request, UploadFile
from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin
from fastapi_users.authentication import (AuthenticationBackend,
                                          BearerTransport, JWTStrategy)
from fastapi_users.db import SQLAlchemyUserDatabase
from src.conf import settings
from src.db.deps import get_user_db
from src.models import User


class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
    reset_password_token_secret = settings.SECRET_KEY
    verification_token_secret = settings.SECRET_KEY
    image_path = "users"

    def __init__(self, user_db: SQLAlchemyUserDatabase, user: Optional[User] = None):
        super().__init__(user_db)
        self.user = user

    @property
    def db_session(self):
        return self.user_db.session

    async def on_after_register(self, user: User, request: Optional[Request] = None):
        logger.debug(f"User {user.id} has registered.")

    async def on_after_forgot_password(self, user: User, token: str, request: Optional[Request] = None):
        logger.debug(f"User {user.id} has forgot their password. Reset token: {token}")

    async def on_after_request_verify(self, user: User, token: str, request: Optional[Request] = None):
        logger.debug(f"Verification requested for user {user.id}. Verification token: {token}")

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

Для аутентификации пользователей используется стратегия JWT (JSON Web Token), которая предоставляет удобный и безопасный способ управления сессиями пользователей. В fastapi-users аутентификация через JWT настраивается следующим образом:

async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)):
    yield UserManager(user_db)


bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")


def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=settings.SECRET_KEY, lifetime_seconds=3600)


auth_backend = AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)

fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])
current_active_user = fastapi_users.current_user(active=True)

Этот код определяет транспортный слой для передачи токенов (Bearer Token) и стратегию их генерации и проверки. JWT-токены генерируются с использованием секретного ключа, хранящегося в настройках приложения, и имеют срок действия, определяемый параметром lifetime_seconds.

Интеграция системы пользователей в FastAPI осуществляется через маршрутизаторы, предоставляемые fastapi-users. Эти маршрутизаторы автоматически создают конечные точки для регистрации, входа, сброса пароля и других операций, связанных с пользователями.

from fastapi import APIRouter
from src.crud import auth_backend, fastapi_users
from src.schemas.user_schema import UserCreate, UserRead, UserUpdate

auth_router = APIRouter()
users_router = APIRouter()

auth_router.include_router(
    fastapi_users.get_auth_router(auth_backend), prefix="/jwt", tags=["auth"]
)
auth_router.include_router(
    fastapi_users.get_register_router(UserRead, UserCreate),
    prefix="",
    tags=["auth"],
)
auth_router.include_router(
    fastapi_users.get_reset_password_router(),
    prefix="",
    tags=["auth"],
)
auth_router.include_router(
    fastapi_users.get_verify_router(UserRead),
    prefix="",
    tags=["auth"],
)

users_router.include_router(
    fastapi_users.get_users_router(UserRead, UserUpdate),
    prefix="",
    tags=["users"],
)

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

Безопасность данных пользователей — одна из важнейших задач при разработке веб-приложений. В этом контексте fastapi-users предоставляет несколько встроенных механизмов для обеспечения безопасности.

1. Хранение паролей: Пароли пользователей никогда не хранятся в открытом виде. Вместо этого fastapi-users использует надежные алгоритмы хеширования для преобразования паролей в безопасные хеши перед их сохранением в базе данных. Это означает, что даже если база данных будет скомпрометирована, злоумышленники не смогут восстановить пароли пользователей.

2. Управление токенами: JWT-токены, используемые для аутентификации, также требуют защиты. Важно использовать достаточно длинные и случайные секретные ключи для их генерации, чтобы предотвратить возможность подделки токенов. Кроме того, следует ограничивать срок действия токенов, чтобы минимизировать риски в случае их утечки.

3. Дополнительные меры безопасности: К дополнительным мерам безопасности относятся:

• Двухфакторная аутентификация (2FA) — добавление второго уровня проверки, что значительно усложняет несанкционированный доступ к учетной записи.
• Мониторинг активности пользователей — отслеживание подозрительных действий и своевременное реагирование на них.
• Использование HTTPS — обязательное шифрование всех данных, передаваемых между клиентом и сервером, что предотвращает перехват данных.

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

Основные принципы:

• Роль — это набор разрешений, который может быть присвоен пользователю. Примеры ролей: admin, editor, viewer.
• Разрешения — конкретные действия, которые может выполнять пользователь с определенной ролью, например, просмотр, редактирование, удаление.
• Иерархия ролей — структура, в которой более привилегированные роли наследуют права от менее привилегированных. Например, роль admin может наследовать все права роли editor.

Для реализации системы ролей в FastAPI можно использовать декораторы и зависимости, которые будут проверять права доступа на уровне маршрутов. Рассмотрим пример настройки и использования ролей.

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

from sqlalchemy import Column, String
from sqlalchemy.orm import Mapped, mapped_column
from src.models.base_model import Base

class User(Base):
    first_name: Mapped[str] = mapped_column(String, nullable=True)
    last_name: Mapped[str] = mapped_column(String, nullable=True)
    role: Mapped[str] = mapped_column(String, default="user")

Здесь роль пользователя хранится в поле role, и по умолчанию всем новым пользователям присваивается роль user.

Следующий шаг — создание зависимости, которая будет проверять роль пользователя при доступе к определенным маршрутам.

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from src.crud.user_crud import current_active_user
from src.models import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/jwt/login")

def get_current_user_role(current_user: User = Depends(current_active_user)) -> str:
    return current_user.role

def role_required(required_role: str):
    def role_dependency(current_role: str = Depends(get_current_user_role)):
        if current_role != required_role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="You do not have the required permissions"
            )
    return role_dependency

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

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

from fastapi import APIRouter, Depends

admin_router = APIRouter()

@admin_router.get("/admin/dashboard", dependencies=[Depends(role_required("admin"))])
async def get_admin_dashboard():
    return {"message": "Welcome to the admin dashboard"}

@admin_router.delete("/admin/user/{user_id}", dependencies=[Depends(role_required("admin"))])
async def delete_user(user_id: str):
    # логика удаления пользователя
    return {"message": f"User {user_id} has been deleted"}

В этом примере доступ к маршрутам /admin/dashboard и /admin/user/{user_id} разрешен только пользователям с ролью admin.

Иерархия ролей позволяет упрощенно управлять правами, назначая более высокие роли, которые наследуют права от более низких. Например, роль admin может включать в себя права ролей editor и viewer.

Для реализации иерархии можно использовать список допустимых ролей и проверять, входит ли текущая роль пользователя в этот список:

def roles_required(*required_roles: str):
    def role_dependency(current_role: str = Depends(get_current_user_role)):
        if current_role not in required_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="You do not have the required permissions"
            )
    return role_dependency

Здесь функция roles_required принимает несколько ролей и проверяет, есть ли текущая роль пользователя среди них. Это позволяет гибко управлять доступом на основе иерархии ролей.

@admin_router.get("/admin/dashboard", dependencies=[Depends(roles_required("admin", "editor"))])
async def get_admin_dashboard():
    return {"message": "Welcome to the admin dashboard"}

@admin_router.delete("/admin/user/{user_id}", dependencies=[Depends(roles_required("admin"))])
async def delete_user(user_id: str):
    # логика удаления пользователя
    return {"message": f"User {user_id} has been deleted"}

В этом примере доступ к /admin/dashboard разрешен как администраторам, так и редакторам, тогда как доступ к /admin/user/{user_id} ограничен только администраторами.

В реальном приложении, где существует множество ролей и сложная система прав доступа, важно поддерживать порядок и четкую структуру. Для этого можно использовать следующие подходы:

• Создание отдельного сервиса для управления ролями: Если приложение масштабируется, имеет смысл вынести управление ролями в отдельный сервис или микросервис, который будет отвечать за централизованное управление ролями и правами.
• Использование групп и разрешений: В некоторых случаях вместо простых ролей стоит использовать группы и разрешения, где группа может включать несколько разрешений, а роли могут быть ассоциированы с одной или несколькими группами.
• Логирование и мониторинг: Важно отслеживать действия пользователей, особенно те, которые требуют повышенных привилегий. Логирование и мониторинг позволят выявить возможные нарушения и быстрее реагировать на них.

HTTPS (HyperText Transfer Protocol Secure) — это расширение протокола HTTP, которое обеспечивает шифрование данных, передаваемых между клиентом и сервером. Использование HTTPS — это обязательный стандарт для всех веб-приложений, особенно если они работают с конфиденциальной информацией, такой как учетные данные пользователей или финансовые данные.

Преимущества использования HTTPS:

• Шифрование данных: Все данные, передаваемые между клиентом и сервером, шифруются, что предотвращает их перехват и модификацию злоумышленниками.
• Проверка подлинности сервера: HTTPS позволяет удостовериться, что клиент подключается к подлинному серверу, а не к поддельному, что защищает от атак типа "человек посередине" (MITM).
• Защита от подделки данных: HTTPS защищает передаваемые данные от изменений и подделки, что особенно важно при передаче конфиденциальной информации.

Для обеспечения HTTPS в FastAPI приложении можно использовать различные подходы, в том числе развертывание приложения за прокси-сервером, таким как Nginx, который будет обрабатывать SSL/TLS сертификаты.

Пример настройки Nginx для FastAPI:

server {
    listen 443 ssl;
    server_name yourdomain.com;

    ssl_certificate /etc/ssl/certs/yourdomain.com.crt;
    ssl_certificate_key /etc/ssl/private/yourdomain.com.key;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

В этом примере Nginx выступает в роли прокси-сервера, который принимает HTTPS-запросы и передает их на FastAPI приложение, работающее на порту 8000.

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

Рекомендации по управлению секретами:

• Храните секреты в защищенных хранилищах: Используйте специальные хранилища секретов, такие как AWS Secrets Manager, HashiCorp Vault или Azure Key Vault, которые обеспечивают безопасное хранение и доступ к конфиденциальным данным.
• Не храните секреты в системе контроля версий: Всегда исключайте файлы, содержащие конфиденциальные данные, из системы контроля версий (например, через .gitignore). Вместо этого используйте файлы конфигурации .env, которые будут загружены только на сервер.
• Используйте переменные окружения: Загружайте секреты в приложение через переменные окружения. Это позволит отделить конфиденциальные данные от кода приложения и упростит управление секретами в различных средах (development, staging, production).

API — это ворота в ваше приложение, и их необходимо защищать от несанкционированного доступа и атак. В FastAPI защита API может включать несколько уровней, начиная от аутентификации и заканчивая защитой от DoS-атак.

Основные меры защиты API:

• Аутентификация и авторизация: Используйте безопасные механизмы аутентификации, такие как JWT или OAuth2, для контроля доступа к API. Убедитесь, что каждый запрос к API проходит проверку прав доступа.
• Ограничение количества запросов (Rate Limiting): Внедрите ограничение количества запросов к API для одного пользователя или IP-адреса, чтобы предотвратить злоупотребления и атаки типа DoS (Denial of Service).
• Валидация и санитизация данных: Проверяйте все входные данные, поступающие в API, на соответствие ожидаемым типам и структурам. Это помогает предотвратить инъекции и другие виды атак, основанных на вводе недопустимых данных.
• Использование HTTPS: Как уже упоминалось, всегда используйте HTTPS для шифрования данных, передаваемых между клиентом и сервером.

Пример настройки ограничения количества запросов с использованием библиотеки fastapi-limiter:

from fastapi import FastAPI, Depends
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter
import aioredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    redis = await aioredis.create_redis_pool("redis://localhost")
    await FastAPILimiter.init(redis)

@app.get("/items", dependencies=[Depends(RateLimiter(times=5, seconds=60))])
async def get_items():
    return {"message": "You can make only 5 requests per minute"}

В этом примере пользователю разрешено делать не более 5 запросов в минуту к маршруту /items.

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

• Обновляйте зависимости: Регулярно проверяйте и обновляйте зависимости вашего проекта. Устаревшие библиотеки могут содержать уязвимости, которые злоумышленники могут использовать для атак.
• Регулярное тестирование безопасности: Проводите регулярные тесты на проникновение (penetration testing) и сканирование уязвимостей, чтобы выявить и устранить слабые места в защите приложения.
• Использование безопасных хешей для паролей: Применяйте надежные алгоритмы хеширования паролей, такие как bcrypt или Argon2, чтобы защитить пароли пользователей в случае утечки базы данных.
• Изолируйте критически важные компоненты: Используйте изоляцию контейнеров и виртуальных машин для защиты критически важных компонентов приложения. Это предотвратит распространение атак в случае компрометации одного из компонентов.
• Журналирование и мониторинг: Настройте систему журналирования и мониторинга, чтобы оперативно отслеживать подозрительную активность и реагировать на инциденты безопасности.

Продолжение следует >>>

PS: Код проекта доступен на GitHub по следующей ссылке: https://github.com/exp-ext/fastapi_template. Обратите внимание, что он может не полностью соответствовать описанному здесь, так как проект находится в процессе развития. В дальнейшем шаблон будет дополняться новыми методами и проходить рефакторинг.