Создание приложений на FastAPI. Часть первая: Введение и подготовка
Введение в FastAPI
Что такое 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
. Виртуальное окружение позволяет изолировать зависимости проекта, что делает его независимым от глобальных настроек системы и предотвращает конфликты между различными проектами.
Создание виртуального окружения:
python3 -m venv venv
Эта команда создаст виртуальное окружение в директории
venv
. После его создания необходимо активировать окружение:Для Linux/MacOS:
source venv/bin/activate
Для Windows:
venv\Scripts\activate
Установка зависимостей: После активации виртуального окружения можно установить все необходимые зависимости. Для этого создается файл
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))
CELERY_BROKER_URL: str = ""
@field_validator("CELERY_BROKER_URL", 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']}//"
)
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="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
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
ASYNC_CELERY_BEAT_DATABASE_URI: PostgresDsn | str = ""
@field_validator("ASYNC_CELERY_BEAT_DATABASE_URI", mode="after")
def assemble_async_celery_beat_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["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 \
gcc \
musl-dev \
postgresql-dev \
libffi-dev
COPY requirements.txt .
RUN pip install --upgrade pip && \
pip install -r requirements.txt --no-cache-dir && \
pip install psycopg2 --no-cache-dir
COPY . .
ENV PYTHONPATH=/app
RUN addgroup -S app-group && \
adduser -S -G app-group app-user && \
chown -R app-user:app-group /app
USER app-user
Разбор Dockerfile
:
Базовый образ:
FROM python:3.11-alpine
Мы используем легковесный образ
python:3.11-alpine
в качестве базового.Alpine
— это минималистичный дистрибутив Linux, который значительно снижает размер итогового Docker-образа.Рабочая директория:
WORKDIR /app
Устанавливаем рабочую директорию
/app
, в которую будут копироваться файлы проекта и где будет выполняться приложение.Установка зависимостей:
RUN apk add --no-cache \ postgresql-client \ gcc \ musl-dev \ postgresql-dev \ libffi-dev
Используем
apk
, пакетный менеджер Alpine, для установкиpostgresql-client
, который необходим для работы с базой данных в контейнере (метод backup и restore db), а остальное для сборкиpsycopg2
.Установка Python-зависимостей:
Сначала копируем файл зависимостей
requirements.txt
в контейнер, затем устанавливаем указанные в нем зависимости. Флаг--no-cache-dir
предотвращает сохранение временных файлов, что уменьшает размер Docker-образа.COPY requirements.txt . RUN pip install -r requirements.txt --no-cache-dir
Копирование файлов проекта:
COPY ./src /app/src COPY ./alembic /app/alembic COPY alembic.ini .
Копируем исходный код приложения, директорию с миграциями Alembic и файл конфигурации Alembic в контейнер.
Создание пользователя и установка прав:
RUN addgroup -S app-group && \ adduser -S -G app-group app-user && \ chown -R app-user:app-group /app USER app-user
Для повышения безопасности создаем группу и пользователя внутри контейнера, затем устанавливаем права на директорию
/app
. После этого переключаемся на созданного пользователя.Настройка переменных окружения (если необходимо):
ENV PYTHONPATH=/app
Устанавливаем переменную окружения
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
:
Общие параметры и переменные окружения:
x-shared-parameters: &shared-parameters env_file: - ../../.env network_mode: "host" restart: always
Использование YAML-анкоров (
&shared-parameters
) и ссылок (<<: *shared-parameters
) позволяет определить общие параметры для всех сервисов, такие как подключение к файлу переменных окружения.env
, режим работы сети и параметры перезапуска.network_mode: "host"
В нашем случае используется исключительно в разработке. В продакшен среде, сеть лучше создавать внутри docker.База данных 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
. Данные базы данных сохраняются на хост-машине в volumetmpl_stage_psgsql_volume
, что обеспечивает их сохранение между запусками контейнеров.Redis:
redis: <<: *shared-parameters image: redis:latest container_name: tmpl_stage_redis command: > --requirepass ${REDIS_PASSWORD} ports: - "6379:6379"
Контейнер с Redis, который будет использоваться для кэширования данных и работы с очередями сообщений. Для повышения безопасности, доступ к Redis защищен паролем.
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
, что позволяет убедиться, что сервис работает корректно перед запуском зависимых от него контейнеров.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
.- Celery и TaskIQ: Контейнеры
celery
,celery-beat
иtaskiq
управляют фоновыми задачами. Они также используют общее окружение и зависимости от базы данных и RabbitMQ. Ограничение памяти и процессорных ресурсов для контейнераcelery-beat
помогает контролировать нагрузку на систему. Volumes:
volumes: tmpl_stage_psgsql_volume: pgadmin_volume: tmpl_stage_rabbitmq_volume:
Использование volumes для хранения данных позволяет контейнерам сохранять свое состояние между перезапусками и обеспечивает безопасность данных.
Параметры сборки и запуска приложения
Для удобства разработки и развертывания проекта используется несколько стратегий:
- Изоляция окружений:
- Использование
.env
файла позволяет настраивать контейнеры в зависимости от окружения (development, production). Это облегчает переключение между различными конфигурациями и поддерживает безопасность, скрывая чувствительные данные.
- Использование
- Удобство разработки:
- Функция
network_mode: "host"
вdocker-compose.yml
обеспечивает прямой доступ к хост-сетям, что упрощает взаимодействие между контейнерами и хост-системой. Это особенно полезно в процессе разработки и тестирования.
- Функция
- Проверка готовности сервисов:
- Использование
depends_on
иhealthcheck
позволяет убедиться, что зависимые сервисы (например, база данных и брокер сообщений) полностью готовы перед запуском основных контейнеров. Это улучшает стабильность системы и предотвращает ошибки запуска.
- Использование
Работа с моделями и CRUD
Проектирование и создание моделей данных
Модели данных лежат в основе любого приложения, взаимодействующего с базой данных. В 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()
Разбор базовой модели:
- Поле
id
:id
— это уникальный идентификатор каждой записи, который представлен в виде UUID. UUID выбирается вместо автоинкрементного целочисленного идентификатора для повышения безопасности и уменьшения риска коллизий при масштабировании приложения.
- Поля
created_at
иupdated_at
:- Эти поля автоматически заполняются при создании и обновлении записи с помощью функций
func.now()
SQLAlchemy. Они полезны для отслеживания времени создания и последнего обновления записей.
- Эти поля автоматически заполняются при создании и обновлении записи с помощью функций
- Метод
__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
:
- Наследование от
SQLAlchemyBaseUserTable
:- Модель
User
наследуется отSQLAlchemyBaseUserTable
, предоставляемогоfastapi-users
. Это позволяет использовать встроенные функции для управления пользователями, такие как хеширование паролей и проверка данных.
- Модель
- Поля
first_name
иlast_name
:- Эти поля хранят имена пользователей и могут быть пустыми (nullable). Они добавляются к основным полям, предоставляемым базовой таблицей пользователей.
- Связи через
relationship
:- Поле
image_files
представляет связь многие-ко-многим между пользователями и изображениями. Используя таблицу ассоциацийuser_image_association
, эта связь управляется автоматически, что упрощает работу с множественными связанными записями. Но об этом в следующей статье.
- Поле
- Метод
__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,
)
Разбор асинхронной конфигурации:
- Асинхронный движок
create_async_engine
:- Движок создается с помощью функции
create_async_engine
, которая позволяет управлять асинхронными соединениями к базе данных. В качестве параметров передаются строка подключения и параметры пула соединений.
- Движок создается с помощью функции
- Асинхронная сессия
async_session
:- Сессия создается с использованием
sessionmaker
и классаAsyncSession
. Это позволяет безопасно управлять транзакциями и асинхронно выполнять запросы к базе данных.
- Сессия создается с использованием
- Пул соединений:
- Использование пулов соединений (
AsyncAdaptedQueuePool
илиNullPool
) позволяет управлять доступом к базе данных в многопоточных приложениях, что улучшает производительность и надежность.
- Использование пулов соединений (
Использование SQLAlchemy в связке с асинхронными базами данных предоставляет разработчикам множество преимуществ:
- Высокая производительность:
- Асинхронные операции позволяют обрабатывать множество запросов одновременно, не блокируя основную логику приложения. Это особенно важно для высоконагруженных систем.
- Удобство и гибкость:
- SQLAlchemy предоставляет гибкий интерфейс для работы с базами данных, позволяя разработчикам сосредоточиться на логике приложения, а не на низкоуровневых деталях работы с SQL.
- Расширяемость:
- Благодаря ORM, модели данных могут быть легко расширены и адаптированы под новые требования. Это упрощает добавление новых функций и поддержание кода.
- Управление связями и транзакциями:
- SQLAlchemy автоматизирует управление сложными связями между таблицами и транзакциями, что уменьшает количество потенциальных ошибок и улучшает целостность данных.
Создание базовых CRUD операций
Веб-приложения зачастую основываются на взаимодействии с базой данных, где необходимо выполнять основные операции: создание, чтение, обновление и удаление (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__)
Разбор кода:
- Параметризация класса:
- Класс
GenericCRUD
параметризован с использованиемTypeVar
для указания типов моделей данных (ModelType
), схем создания (CreateSchemaType
) и схем обновления (UpdateSchemaType
). Это позволяет использовать один и тот же класс для работы с различными моделями, сохраняя при этом строгую типизацию и безопасность.
- Класс
- Типизация:
- Типизация играет ключевую роль в обеспечении корректности кода. Использование типовых переменных позволяет IDE и инструментам статического анализа кода (таким как
mypy
) проверять соответствие типов и предотвращать ошибки на этапе компиляции, что значительно повышает надежность приложения.
- Типизация играет ключевую роль в обеспечении корректности кода. Использование типовых переменных позволяет IDE и инструментам статического анализа кода (таким как
Рассмотрим методы класса 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
типизация позволяет:
- Снизить количество ошибок:
- IDE и линтеры могут автоматически выявлять несоответствия типов, что снижает вероятность ошибок в коде.
- Улучшить читаемость и поддержку кода:
- Явная типизация делает код более понятным и поддерживаемым, так как разработчики сразу видят, с какими типами данных работают методы.
- Повысить надежность и предсказуемость:
- Типизированные методы и классы позволяют разработчикам быть уверенными, что данные, передаваемые между различными частями приложения, соответствуют ожидаемым структурам и типам.
Кэширование
Кэширование — это эффективный способ оптимизации веб-приложений, который позволяет значительно снизить нагрузку на сервер, уменьшив количество повторяющихся запросов к базе данных или внешним API. В FastAPI кэширование можно легко реализовать с помощью библиотеки fastapi-cache2
и Redis.
Преимущества кэширования:
- Уменьшение времени отклика:
- Данные, сохраненные в кэше, могут быть извлечены значительно быстрее, чем из базы данных. Это особенно важно для данных, которые редко изменяются, но часто запрашиваются.
- Снижение нагрузки на сервер:
- За счет кэширования уменьшается количество запросов к базе данных, что позволяет уменьшить нагрузку на сервер и повысить его производительность.
- Повышение устойчивости приложения:
- В случае временной недоступности базы данных или внешнего 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-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. Обратите внимание, что он может не полностью соответствовать описанному здесь, так как проект находится в процессе развития. В дальнейшем шаблон будет дополняться новыми методами и проходить рефакторинг.