Перейти к содержанию

Зависимости сервисного слоя

Модуль dependencies предоставляет фабрики и провайдеры зависимостей для сервисного слоя приложения, используя паттерн внедрения зависимостей.

Особенности

  • Кэширование часто используемых зависимостей
  • Асинхронная загрузка зависимостей
  • Поддержка единицы работы (Unit of Work)
  • Интеграция с репозиториями и сервисами
  • Управление жизненным циклом зависимостей

Документация API

get_settings

Фабрика для получения настроек приложения с кэшированием.

Source code in src/service_layer/dependencies.py 
15
16
17
@lru_cache
def get_settings() -> Settings:
    return Settings()  # type: ignore

get_uow

Провайдер для единицы работы (Unit of Work).

Source code in src/service_layer/dependencies.py 
20
21
22
23
24
async def get_uow() -> AbstractUnitOfWork:
    return SqlAlchemyUnitOfWork(
        session_factory=get_session_factory(),
        repo_factory=await get_repo_factory(),
    )

get_repo_factory

Фабрика для создания репозиториев.

Source code in src/service_layer/dependencies.py 
27
28
async def get_repo_factory() -> ABCUsersRepositoryFactory:
    return SQLAlchemyUsersRepositoryFactory(await get_hasher())

get_hasher

Провайдер для сервиса хеширования паролей.

Source code in src/service_layer/dependencies.py 
31
32
async def get_hasher() -> IPasswordHasher:
    return UserPasswordHasher()

get_user_service

Фабрика для сервиса работы с пользователями.

Source code in src/service_layer/dependencies.py 
35
36
37
38
async def get_user_service() -> UserService:
    uow = await get_uow()
    hasher = await get_hasher()
    return UserService(uow=uow, hasher=hasher)

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

Базовое использование в FastAPI

from fastapi import APIRouter, Depends
from src.service_layer.dependencies import get_user_service
from src.service_layer.users_service import UserService

router = APIRouter()

@router.get("/users/{user_id}")
async def get_user(
    user_id: int,
    user_service: UserService = Depends(get_user_service)
):
    user = await user_service.get_user(user_id)
    return {"user": user}

Кастомизация зависимостей

from functools import lru_cache
from src.service_layer.dependencies import get_settings, get_hasher
from src.adapters.factory import CustomUsersRepositoryFactory

# Переопределение фабрики репозиториев
async def get_custom_repo_factory():
    hasher = await get_hasher()
    return CustomUsersRepositoryFactory(hasher)

# Использование кастомной фабрики
@lru_cache
def get_custom_settings():
    settings = get_settings()
    # Модификация настроек
    settings.DEBUG = True
    return settings

Рекомендации по использованию

  1. Управление зависимостями
  2. Используйте lru_cache для дорогих в создании зависимостей
  3. Разделяйте зависимости по уровням абстракции

  4. Избегайте циклических импортов

  5. Тестирование

  6. Заменяйте зависимости на моки в тестах
  7. Используйте фикстуры для настройки тестового окружения

  8. Изолируйте тесты друг от друга

  9. Производительность

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

Интеграция

Модуль интегрируется с:

  • FastAPI приложениями
  • SQLAlchemy ORM
  • Системой аутентификации
  • Сервисным слоем приложения
  • Репозиториями доступа к данным

Ограничения

  • Требуется Python 3.8+
  • Зависит от FastAPI и SQLAlchemy
  • Предполагает использование асинхронного кода

Дополнительные возможности

Создание кастомных провайдеров

from typing import Any
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db_session() -> AsyncSession:
    """Провайдер сессии БД с автоматическим управлением жизненным циклом."""
    session_factory = get_session_factory()
    async with session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# Использование в зависимостях
async def get_user_service(
    session: AsyncSession = Depends(get_db_session)
) -> UserService:
    return UserService(session=session)

Кэширование зависимостей

from functools import lru_cache
from src.service_layer.dependencies import get_settings

@lru_cache
def get_cached_service():
    """Пример кэширования сервиса с настройками."""
    settings = get_settings()
    return SomeService(settings)

# Использование в FastAPI маршруте
@router.get("/some-route")
async def some_route(
    service: SomeService = Depends(get_cached_service)
):
    result = await service.do_something()
    return {"result": result}