Test Service

Core модуль, который дает единые интерфейсы для работы с БД и Redis. Остальные модули могут использовать их через Dependency Injection.

Запуск

Через Docker (самый простой способ)

# Копируем настройки
cp .env.example .env

# В .env меняем хосты для docker:
# POSTGRES_HOST=postgres
# REDIS_HOST=redis

# Запускаем всё
docker compose up -d

Приложение будет на http://localhost:8000

Локально (для разработки)

# Ставим зависимости
poetry install

# Поднимаем только БД и Redis
docker compose up -d postgres redis

# В .env используем localhost:
# POSTGRES_HOST=localhost
# REDIS_HOST=localhost

# Запускаем приложение
poetry run uvicorn app.main:app --reload

Как использовать

Работа с базой данных

Просто добавляешь DatabaseDep в параметры роута:

from fastapi import APIRouter
from sqlalchemy import select
from app.core.database import DatabaseDep

router = APIRouter()

@router.get("/users")
async def get_users(db: DatabaseDep):
    result = await db.execute(select(User))
    return result.scalars().all()

Сессия автоматически закроется после запроса, транзакция закоммитится (или откатится при ошибке).

Работа с Redis

Аналогично, через RedisDep:

from app.core.redis import RedisDep

@router.get("/cache/{key}")
async def get_cache(key: str, redis: RedisDep):
    value = await redis.get(key)
    return {"key": key, "value": value}

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

Если нужно что-то сделать вне роутов:

from app.core.database import db_manager
from app.core.redis import redis_manager

# Работа с БД
async with db_manager.get_session() as session:
    # делаешь что нужно
    await session.execute(...)
    await session.commit()

# Работа с Redis
client = redis_manager.client
await client.set("key", "value")

Endpoints

• GET / - просто проверка, что всё работает
• GET /health - статус подключений к БД и Redis
• GET /docs - Swagger документация

Архитектурные решения

Разделение ответственности

Каждый модуль отвечает за свою область:

• config.py - только конфигурация, загрузка из .env
• database.py - только работа с PostgreSQL
• redis.py - только работа с Redis
• main.py - только инициализация FastAPI приложения

Это упрощает тестирование и поддержку - изменения в одном модуле не затрагивают другие.

Применение паттернов проектирования

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

Dependency Injection - через FastAPI Depends. Каждый запрос получает свою сессию БД, которая автоматически закрывается после обработки. Не нужно вручную управлять жизненным циклом.

Context Manager - для сессий БД. Автоматический commit при успехе, rollback при ошибке, гарантированное закрытие сессии.

Lifespan Events - инициализация подключений при старте приложения, корректное закрытие при остановке

Factory Method - async_sessionmaker создает сессии по требованию, aioredis.from_url() создает клиент с нужными настройками.

Управление ресурсами

• Connection pooling - пул соединений для PostgreSQL (настраивается через postgres_pool_size)
• Автоматическое закрытие - все подключения закрываются через lifespan при остановке
• Health checks - проверка доступности Redis через health_check()
• Pool pre-ping - проверка соединений перед использованием (избегаем "connection lost")
• Pool recycle - переиспользование соединений через час

Организация настроек и валидация

Используется Pydantic Settings:

• Автоматическая загрузка из .env
• Валидация типов при старте (не во время выполнения)
• Валидация значений (например, environment может быть только development/staging/production)
• Типобезопасность - IDE подсказывает доступные параметры
• Значения по умолчанию для всех параметров

Обработка ошибок и граничных случаев

• Транзакции БД - автоматический rollback при любом исключении
• Проверка инициализации - RuntimeError если пытаешься использовать неинициализированный менеджер
• Health checks - graceful degradation, приложение может работать даже если Redis недоступен
• Connection retry - Redis клиент настроен на retry_on_timeout
• Валидация конфигурации - ошибки конфигурации видны сразу при старте, а не в runtime

Реализация lifecycle приложения

Используется lifespan context manager:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await db_manager.initialize()
    await redis_manager.initialize()
    yield
    # Shutdown
    await db_manager.close()
    await redis_manager.close()

Это гарантирует:

• Подключения инициализируются до обработки запросов
• Подключения закрываются корректно при остановке
• Graceful shutdown - все ресурсы освобождаются

Dependency Injection

Реализовано через FastAPI Depends:

• get_db() - async генератор, возвращает сессию БД
• get_redis() - async генератор, возвращает Redis клиент
• Типы DatabaseDep и RedisDep для удобства использования

Преимущества:

• Автоматическое управление жизненным циклом
• Легко тестировать (можно подменить через app.dependency_overrides)
• Изоляция между запросами
• Чистый код без явного управления ресурсами

Возможность расширения

Добавление моделей SQLAlchemy

from app.core.database import Base
from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)

Добавление новых роутов

Просто создай новый роутер и используй DatabaseDep/RedisDep:

from fastapi import APIRouter
from app.core.database import DatabaseDep
from app.core.redis import RedisDep

router = APIRouter(prefix="/api/v1")

@router.get("/items")
async def get_items(db: DatabaseDep, redis: RedisDep):
    # используй db и redis как нужно
    pass

Расширение Redis функционала

Можно добавить методы-хелперы в RedisManager:

• Кэширование с TTL
• Pub/Sub для событий
• Транзакции
• Lua скрипты

Миграции Alembic

Для управления схемой БД:

poetry add alembic
alembic init alembic
# в alembic.ini укажи sqlalchemy.url = settings.postgres_url

Идеи по улучшению

1. Connection Retry Logic - автоматические повторы при потере соединения с экспоненциальной задержкой
2. Metrics - добавить Prometheus метрики для мониторинга пула соединений и времени ответа
3. Logging - структурированное логирование с контекстом (request_id, user_id и т.д.)
4. Read Replicas - поддержка чтения из реплик PostgreSQL для распределения нагрузки
5. Redis Sentinel/Cluster - поддержка Redis Sentinel или Cluster для высокой доступности
6. Circuit Breaker - паттерн Circuit Breaker для устойчивости к сбоям внешних сервисов
7. Graceful Degradation - более умная обработка недоступности Redis (кэш опционален)
8. Connection Pool Metrics - мониторинг использования пула (сколько соединений занято/свободно)
9. Type Safety - более строгая типизация, TypedDict для конфигурации
10. Testing Utilities - фикстуры для pytest, тестовые контейнеры для БД/Redis

Замечания

• При первом запуске может потребоваться время на создание базы данных
• Health check для PostgreSQL не реализован
• В production стоит добавить более строгие настройки пула соединений
• Для продакшена лучше использовать секреты из vault, а не .env файлы
• Стоит добавить rate limiting для защиты от перегрузки