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

Рецепты и полезные советы

В этом разделе собраны практические примеры решения типичных задач с помощью chutils.

1. Логирование

Асинхронное логирование (Performance)

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

from chutils import setup_logger

# Включение асинхронного режима
logger = setup_logger(use_async=True)

# Теперь основной поток не будет ждать завершения записи на диск
logger.info("Это сообщение будет обработано в фоновом потоке")

Расширенное маскирование PII

chutils может автоматически скрывать чувствительные данные (email, карты) не только по конкретным значениям, но и по паттернам.

from chutils import setup_logger

# Настройка автоматического маскирования email и телефонов
logger = setup_logger(
    use_predefined_patterns=["email", "phone"],
    custom_patterns=[r"ID-\d{4}"]  # Свои регулярные выражения
)

# Выведет: "Contact user [MASKED] at [MASKED]"
logger.info("Contact user ID-1234 at test@example.com")

Несколько логгеров для разных модулей

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

from chutils import setup_logger

# Логгер для сетевого модуля (только ошибки в файл network.log)
net_logger = setup_logger("network", log_level="ERROR", log_file_name="network.log")

# Логгер для ядра (детальная отладка в консоль и core.log)
core_logger = setup_logger("core", log_level="DEVDEBUG", log_file_name="core.log")

net_logger.error("Ошибка соединения!")
core_logger.devdebug("Состояние объекта: %s", obj_data)

Настройка через конфигурационный файл

Вы можете централизованно управлять всеми логгерами в config.yml:

# config.yml
Logging:
  log_level: INFO
  rotation_type: time

AuditLogger:
  log_level: DEBUG
  log_file_name: "audit.log"

В коде:

# Использует секцию AuditLogger
audit_logger = setup_logger("audit", config_section_name="AuditLogger")

Предотвращение конфликтов конфигурации логов (pydantic-settings)

[!WARNING] Если ваше приложение одновременно использует chutils (для логирования и чтения config.yml) и pydantic-settings ( для загрузки настроек, таких как LOG_LEVEL из .env / системного окружения), у вас работают две независимые системы конфигурации.

setup_logger_from_config() читает уровень логирования из секции Logging.log_level в config.yml, тогда как pydantic-settings считывает LOG_LEVEL из .env. Они никак не связаны по умолчанию. Изменение уровня логов в .env не повлияет на логгер chutils.

Для решения этого конфликта рекомендуется выбрать один из двух подходов:

Вариант А. Единый источник правды через chutils (Рекомендуется)

Полностью доверьтесь механизму конфигурации chutils (config.yml и config.local.yml). Удалите log_level из вашей Pydantic-модели и .env файла. Управляйте уровнем логирования в config.yml, а для локальной разработки переопределяйте его в config.local.yml (или используйте переменные окружения CH_LOGGING_LOG_LEVEL).

В этом случае вы просто вызываете логгер без аргументов:

from chutils import setup_logger_from_config

# Настройки уровня берутся из config.yml (секция Logging)
logger = setup_logger_from_config()

Вариант Б. Явный мост (Экспорт из Pydantic)

Если вы хотите продолжать хранить LOG_LEVEL в .env и управлять им через Pydantic-модель, явно передавайте это значение при настройке логгера:

from pydantic_settings import BaseSettings
from chutils import setup_logger


class AppSettings(BaseSettings):
    log_level: str = "INFO"

    class Config:
        env_file = ".env"


# Инициализируем настройки приложения (читает .env и системное окружение)
settings = AppSettings()

# Явно связываем настройки Pydantic и логгер chutils
logger = setup_logger(log_level=settings.log_level)

Контекстное логирование в FastAPI / asyncio

Если вы хотите автоматически добавлять ID запроса во все логи без передачи его через аргументы функций.

from chutils import setup_logger, bind_context
import asyncio

logger = setup_logger()


async def deep_nested_function():
    # Нам не нужно передавать request_id сюда, он подхватится сам!
    logger.info("Лог из глубины приложения")


async def handle_request(request_id: str):
    bind_context(request_id=request_id)
    logger.info("Начало обработки")
    await deep_nested_function()


# В асинхронном цикле контексты изолированы
asyncio.gather(
    handle_request("REQ-1"),
    handle_request("REQ-2")
)

2. Кэширование (Smart Caching)

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

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

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

  1. Полного имени функции (включая модуль).
  2. Всех позиционных аргументов (args).
  3. Всех именованных аргументов (kwargs).

Это значит, что вызовы с разными данными будут кэшироваться независимо:

@cache_with_ttl(ttl=60)
def calculate_price(item_id: int, discount: float = 0.0):
    # Вычисления...
    return final_price


# Разные аргументы = разные записи в кэше
calculate_price(1)  # Вычислится и сохранится для key_1
calculate_price(2)  # Вычислится и сохранится для key_2
calculate_price(1, 0.1)  # Вычислится и сохранится для key_3
calculate_price(1)  # Вернется из кэша (мгновенно)

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

import asyncio

from chutils.cache import cache_with_ttl


# Кэшируем результат на 60 секунд
@cache_with_ttl(ttl=60)
def get_heavy_data(user_id: int):
    print(f"Вычисляем данные для {user_id}...")
    return {"id": user_id, "data": "..."}


# Асинхронный кэш с префиксом ключа
@cache_with_ttl(ttl=300, key_prefix="api_response")
async def fetch_remote_api(url: str):
    await asyncio.sleep(1)  # Имитация задержки
    return {"url": url, "status": "ok"}

Инвалидация и тегирование кэша

Если данные во внешнем источнике изменились, вы можете принудительно сбросить закэшированные значения:

1. Точечная инвалидация по аргументам

Удаляет запись только для конкретного набора аргументов, с которыми вызывалась функция:

@cache_with_ttl(ttl=300)
def get_user_profile(user_id: int):
    return {"id": user_id, "name": "Alice"}


# Сбросит кэш только для вызова get_user_profile(42)
get_user_profile.invalidate(42)

# Для асинхронных функций:
# await get_user_profile.ainvalidate(42)

2. Очистка всего кэша функции

Сбрасывает все сохраненные записи кэша, сгенерированные данной функцией:

get_user_profile.invalidate_all()

# Для асинхронных функций:
# await get_user_profile.ainvalidate_all()

3. Инвалидация по тегам (Cache Tagging)

Вы можете группировать кэш-записи с помощью тегов. Теги могут быть как статическими (список строк), так и динамическими (callable, принимающий те же аргументы, что и функция):

# Использование динамического тега
@cache_with_ttl(ttl=300, tags=lambda user_id: [f"user_{user_id}", "all_users"])
def get_user_permissions(user_id: int):
    return ["read", "write"]


# Сбросит кэш разрешений только для пользователя 42
get_user_permissions.invalidate_tag("user_42")

# Сбросит кэш разрешений для всех пользователей
get_user_permissions.invalidate_tag("all_users")

# Для асинхронных функций:
# await get_user_permissions.ainvalidate_tag("all_users")

3. Работа с конфигурацией

Использование относительных путей

chutils умеет автоматически делать пути абсолютными относительно корня проекта.

# config.yml
Paths:
  upload_dir: "data/uploads"

В коде:

from chutils import get_config_path

# Если корень /home/user/project, вернет /home/user/project/data/uploads
upload_path = get_config_path("Paths", "upload_dir")

Локальные переопределения

Создайте config.local.yml (и добавьте его в .gitignore), чтобы переопределить настройки для разработки:

# config.local.yml
Database:
  host: "localhost" # На сервере будет production.db.com

Специфичные для окружения конфигурации

Вы можете создавать отдельные файлы настроек для разных сред развертывания (например, staging, production). Библиотека автоматически подхватит нужный файл на основе переменной окружения CH_ENV.

  1. Создайте файл config.production.yml.
  2. Установите CH_ENV=production в вашей среде.

Приоритет загрузки (от высшего к низшему):

  1. Переменные окружения (CH_SECTION_KEY).
  2. Локальный файл (config.local.yml).
  3. Файл окружения (config.{CH_ENV}.yml).
  4. Основной файл (config.yml).

4. Управление секретами

Использование в Docker / CI-CD

В изолированных средах системное хранилище (Keyring) часто недоступно. Чтобы избежать лишних предупреждений и ошибок:

  1. Установите переменную окружения CH_DISABLE_KEYRING_WARNING=true.
  2. Используйте .env файл для хранения секретов.
# .env
DB_PASSWORD="my-safe-password"

SecretManager автоматически подхватит это значение.

5. Hot-Reload конфигурации

Автоматическое обновление состояния приложения

Если ваше приложение должно менять свое поведение (например, уровень логирования или лимиты) без перезагрузки.

from chutils import (
    setup_logger,
    start_config_watcher,
    on_config_change,
    get_config_value
)

logger = setup_logger()


def update_app_state():
    # Читаем новые значения
    new_limit = get_config_value("App", "rate_limit", 100)
    logger.info(f"Лимит обновлен: {new_limit}")

    # Здесь можно обновить объект приложения или глобальное состояние
    # app.rate_limiter.set_limit(new_limit)


# 1. Подписываемся на изменения
on_config_change(update_app_state)

# 2. Запускаем мониторинг
start_config_watcher()

Использование с Pydantic моделями

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

def on_reload():
    # При вызове заново будет создана новая модель с актуальными данными
    cfg = get_config(model=AppConfig)
    print(f"Новое имя приложения: {cfg.name}")

6. Декораторы

Отладка производительности

Используйте декоратор @log_function_details для быстрого анализа работы функций без изменения их кода.

from chutils import log_function_details, setup_logger

# Важно: установите уровень DEVDEBUG, чтобы увидеть вывод декоратора
setup_logger(log_level="DEVDEBUG")


@log_function_details
def process_heavy_task(data):
    # Какая-то логика...
    return True


process_heavy_task([1, 2, 3])

В логах появится время выполнения с точностью до миллисекунд и переданные аргументы.

7. Валидация через Pydantic

Строгая типизация всей конфигурации

Вы можете описать ожидаемую структуру вашего config.yml в виде Pydantic моделей для автоматической валидации при загрузке.

from pydantic import BaseModel, Field
from chutils import get_config


class DbConfig(BaseModel):
    host: str
    port: int


class AppConfig(BaseModel):
    name: str
    version: str
    db: DbConfig = Field(alias="Database")


# Валидация и автодополнение
cfg = get_config(model=AppConfig)
print(f"Подключение к {cfg.db.host}:{cfg.db.port}")

Валидация отдельной секции

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

from chutils import get_config_section

db_cfg = get_config_section("Database", model=DbConfig)

Групповая валидация обязательных ключей

Если вам нужно проверить наличие нескольких обязательных переменных конфигурации (например, при запуске сервиса) без использования тяжелых Pydantic моделей, используйте validate_required_keys. В случае отсутствия одного или нескольких ключей функция сгенерирует агрегированную ошибку ConfigValidationGroupError.

from chutils.config import validate_required_keys
from chutils.exceptions import ConfigValidationGroupError

try:
    validate_required_keys(
        section="Secrets",
        keys=["telegram_bot_token", "database_url", "api_key"]
    )
except ConfigValidationGroupError as e:
    # Выведет структурированный список всех недостающих настроек за один проход
    print(f"Ошибка инициализации сервиса: {e}")

Интеграция с IDE (VSCode, PyCharm) через JSON Schema

Чтобы получить автодополнение и проверку типов прямо в YAML файле, вы можете сгенерировать JSON Schema для вашей Pydantic модели и подключить её.

  1. Генерация схемы: bash chutils config generate-schema --model my_app.models:Settings -o .config.schema.json

  2. Подключение в VSCode: Добавьте "магический комментарий" в начало вашего config.yml: ```yaml # yaml-language-server: $schema=./.config.schema.json

Database: host: localhost port: 5432 ``` Требуется расширение "YAML" от Red Hat.

  1. Подключение в PyCharm:
    • Перейдите в Settings -> Languages & Frameworks -> Schemas and DTDs -> JSON Schema Mappings.
    • Добавьте новую схему, укажите путь к .config.schema.json и выберите ваш файл config.yml.

8. Утилита командной строки (CLI)

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

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

# Сохранить API ключ
chutils secrets set STRIPE_KEY "sk_test_..."

# Удалить секрет
chutils secrets delete STRIPE_KEY

# Указать конкретный сервис (по умолчанию - имя текущей папки)
chutils secrets set AWS_SECRET "..." --service my-production-app

9. Работа с файловой системой

Безопасная запись данных

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

from chutils.fs import atomic_write

config_data = {"version": 1, "settings": {"theme": "dark"}}

# Автоматически сериализует в JSON, так как расширение .json
atomic_write("settings.json", config_data)

# Автоматически сериализует в YAML, так как расширение .yaml
atomic_write("settings.yaml", config_data)

# Запись обычного текста
atomic_write("hello.txt", "Hello world")

Временные файлы с авто-удалением

Контекстный менеджер get_temp_file создает временный файл и гарантированно удаляет его при выходе из блока with.

from chutils.fs import get_temp_file

with get_temp_file(suffix=".tmp") as temp_path:
    # Делаем что-то с временным файлом
    temp_path.write_text("temporary content")
    print(f"Путь к файлу: {temp_path}")

# Здесь файл уже удален

10. Graceful Shutdown (Управление жизненным циклом)

Механизм корректного завершения работы приложения позволяет выполнить необходимые действия по очистке ресурсов (закрытие соединений с БД, логов, сокетов) при получении сигналов от ОС (например, Ctrl+C).

Регистрация функций очистки

Используйте декоратор @register_cleanup для регистрации как синхронных, так и асинхронных функций.

import asyncio
from chutils import register_cleanup, setup_graceful_shutdown


@register_cleanup
async def close_db():
    print("Closing database connections...")
    await asyncio.sleep(0.1)  # Имитация работы
    print("DB connections closed.")


@register_cleanup
def cleanup_temp_files():
    print("Deleting temporary files...")


# В начале работы приложения активируйте перехват сигналов
setup_graceful_shutdown()


# Пример работы асинхронного цикла
async def main():
    print("App is running... Press Ctrl+C to stop.")
    while True:
        await asyncio.sleep(1)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Настройка таймаута

По умолчанию на выполнение всех функций очистки дается 10 секунд. Вы можете изменить это значение в config.yml:

# config.yml
shutdown:
  timeout: 5 # Таймаут в секундах

Особенности работы

  1. LIFO (Last-In-First-Out): Функции выполняются в обратном порядке их регистрации. Это удобно, если ресурсы зависят друг от друга (например, сначала закрыть логгер, потом БД).
  2. Log and Continue: Если одна из функций выбросит исключение, chutils залогирует ошибку и продолжит выполнение остальных функций.
  3. Кроссплатформенность: На Windows перехватываются SIGINT и SIGTERM, на Linux/Unix дополнительно SIGHUP.

11. Работа со временем (Painless Datetime)

Модуль chutils.time обеспечивает "UTC-first" подход, гарантируя, что вы всегда работаете с осведомленными (timezone aware) объектами времени.

Получение текущего времени в UTC

from chutils import utc_now

# Возвращает datetime с tzinfo=timezone.utc
now = utc_now()
print(f"Текущее время: {now}")

Умный парсинг дат

Функция parse_datetime поддерживает ISO строки, UNIX-таймстампы (в секундах и миллисекундах) и автоматически приводит их к UTC.

from chutils import parse_datetime

# ISO 8601
dt1 = parse_datetime("2023-10-27T12:00:00")

# UNIX Timestamp (секунды)
dt2 = parse_datetime(1698400000)

# UNIX Timestamp (миллисекунды)
dt3 = parse_datetime(1698400000000)

# Если установлена библиотека chutils[date], поддерживается любой формат:
# dt4 = parse_datetime("27 Oct 2023 12:00")

Человекочитаемая разница во времени

Превращает разницу между датами в понятные строки на русском или английском языке.

from datetime import timedelta
from chutils import utc_now, humanize_timedelta

now = utc_now()
past_date = now - timedelta(minutes=5)
future_date = now + timedelta(days=1)

print(humanize_timedelta(past_date))  # "5 минут назад"
print(humanize_timedelta(future_date))  # "завтра"
print(humanize_timedelta(past_date, locale='en'))  # "5 minutes ago"

12. Быстрое создание CLI (CLI Booster)

Декоратор @cli_command позволяет превратить любую функцию в полноценный CLI-инструмент за одну секунду. Он автоматически создает парсер аргументов на основе сигнатуры функции.

Простой скрипт

# my_tool.py
from pathlib import Path

from chutils import cli_command


@cli_command
def copy_files(source: Path, dest: Path, verbose: bool = False):
    """
    Утилита для копирования файлов.

    Args:
        source (Path): Путь к исходному файлу.
        dest (Path): Путь назначения.
        verbose (bool): Выводить подробную информацию.
    """
    if verbose:
        print(f"Копируем из {source} в {dest}")
    # Логика...


if __name__ == "__main__":
    copy_files()

Теперь вы можете запустить его из терминала:

python my_tool.py /tmp/src /tmp/dst --verbose
python my_tool.py --help

Асинхронные команды и списки

CLI Booster отлично справляется с асинхронностью и списками аргументов.

import asyncio
from chutils import cli_command


@cli_command
async def process_batch(ids: list[int], retry: int = 3):
    """Обработка списка ID с повторами."""
    for item_id in ids:
        print(f"Processing {item_id} (retries: {retry})")
        await asyncio.sleep(0.1)


if __name__ == "__main__":
    process_batch()

13. Отладка и диагностика конфигурации (Config Diagnostics)

Если вы не понимаете, почему значение ключа в приложении отличается от того, что написано в config.yml, используйте интерактивный отладчик.

Использование через CLI

Команда config debug покажет всю историю изменений для каждого ключа: откуда он был загружен изначально и чем перекрыт позже.

# Показать дерево конфигурации (по умолчанию)
chutils config debug

# Вывод в виде таблицы
chutils config debug --format table

# Показать секретные значения (по умолчанию они маскируются)
chutils config debug --show-secrets

# Экспорт в JSON для анализа
chutils config debug --format json > config_trace.json

Использование через API

Вы можете включить трассировку и получить данные программно:

from chutils.config.manager import _cm
from chutils.config import get_config
from chutils.config.diagnostics import format_trace

# 1. Включаем сбор метаданных
_cm.tracing_enabled = True

# 2. Сбрасываем кэш и загружаем конфиг
_cm.clear_cache()
get_config()

# 3. Получаем и форматируем отчет
trace = _cm.get_trace()
print(format_trace(trace, format_type='tree'))

14. Распределенное трассирование (OpenTelemetry)

chutils предоставляет легковесную интеграцию с OpenTelemetry для отслеживания пути выполнения запросов и связи логов с трассами.

Установка

Функционал трассировки является опциональным:

pip install chutils[otel]

Быстрый старт

  1. Настройте сбор трасс в начале вашего приложения:
from chutils import setup_tracing

# Настройка вывода трасс в консоль (для локальной разработки)
setup_tracing(service_name="my_service", exporter_type="console")
  1. Используйте декоратор @trace для функций, которые хотите отслеживать:
from chutils import trace, setup_logger

logger = setup_logger()


@trace(capture_kwargs=True)
def process_order(order_id: int):
    logger.info("Начинаем обработку заказа")
    # ... логика ...
    return True

Особенности

  • Связь с логами: В текстовых логах автоматически появятся [trace_id=... span_id=...]. В JSON логах эти поля будут вынесены на верхний уровень.
  • Async: Декоратор @trace полностью поддерживает асинхронные функции.
  • OTLP: Для промышленного использования (Jaeger, Grafana Tempo) используйте exporter_type="otlp".
  • Zero Overhead: Если пакеты opentelemetry не установлены, декоратор @trace не создает никаких накладных расходов.

15. Дистанционная конфигурация (Remote Config)

chutils позволяет загружать настройки из удаленных HTTP/HTTPS источников. Это полезно для централизованного управления конфигурациями в микросервисной архитектуре.

Быстрый старт

Просто укажите URL при получении конфигурации:

from chutils import get_config

# Загрузка и объединение с локальными файлами
config = get_config(remote_url="https://api.example.com/config.json")

Периодический опрос (Polling)

Вы можете настроить автоматическое фоновое обновление конфигурации:

# Опрос каждые 60 секунд
config = get_config(
    remote_url="https://api.example.com/config.json",
    polling_interval=60
)

Динамический интервал

Вы можете управлять интервалом опроса прямо из удаленного конфига. Если в загруженных данных есть секция RemoteConfig (или polling) с ключом interval, chutils автоматически переключится на этот интервал.

{
  "RemoteConfig": {
    "interval": 300
  },
  "Database": {
    "host": "remote-db"
  }
}

Авторизация и безопасность

Для доступа к защищенным эндпоинтам используйте remote_auth:

config = get_config(
    remote_url="https://secure-config.local/app.yml",
    remote_auth=("admin", "secret-token")
)

Отказоустойчивость (Fallback)

Если удаленный сервер временно недоступен, chutils автоматически вернет последнюю успешно загруженную версию из памяти (кэша), чтобы приложение продолжало работать.

16. Инструменты разработчика и AI-контекст

Если вы хотите быстро создать документацию по API вашего проекта или подготовить глубокий индекс для AI-агента.

Генерация карты API

Создает Markdown-файл со списком всех публичных функций, классов и их описаний.

chutils dev generate-context -o api_map.md

Генерация семантического индекса для AI

Генерирует JSON-дерево проекта (через AST), которое включает связи между модулями, веса зависимостей и метаданные символов. Это "золотой стандарт" контекста для современных LLM.

chutils dev generate-context --tree -o project_index.json

17. Внутренняя шина событий (In-Memory Event Bus)

Шина событий (chutils.events) позволяет развязать компоненты вашего приложения (loose coupling) через паттерн Pub/Sub.

Подписка и публикация

Используйте декоратор @subscribe для регистрации обработчиков и publish (или publish_async) для отправки событий.

from chutils.events import subscribe, publish, publish_async
import asyncio


# 1. Синхронный обработчик
@subscribe("user_created")
def send_email(user_id: int, username: str):
    print(f"Отправка письма для {username}...")


# 2. Асинхронный обработчик
@subscribe("user_created")
async def setup_profile(user_id: int, username: str):
    await asyncio.sleep(0.1)
    print(f"Профиль {username} настроен.")


# Публикация событий:
# Синхронная (запускает асинхронных подписчиков в фоновом режиме):
publish("user_created", user_id=1, username="Дмитрий")

# Асинхронная (ждет завершения всех подписчиков):
# await publish_async("user_created", user_id=2, username="Анна")

Стратегии обработки ошибок

При возникновении исключений в обработчиках вы можете выбрать одну из трех стратегий:

  1. IGNORE (по умолчанию) — логирует ошибки через chutils.logger и продолжает работу остальных обработчиков.
  2. FAIL_FAST — немедленно прерывает выполнение и пробрасывает ошибку.
  3. COLLECT — выполняет все задачи, собирает ошибки и выбрасывает объединенное исключение EventBusExceptionGroup.
from chutils.events import EventBus, ErrorStrategy
from chutils.exceptions import EventBusExceptionGroup

bus = EventBus(error_strategy=ErrorStrategy.COLLECT)


@bus.subscribe("data_sync")
def step_1():
    raise ValueError("Сбой шага 1")


@bus.subscribe("data_sync")
def step_2():
    raise IOError("Сбой шага 2")


try:
    bus.publish("data_sync")
except EventBusExceptionGroup as e:
    # Классический перехват (совместим со всеми версиями Python >= 3.10)
    print(f"Возникли ошибки: {e.exceptions}")

# На Python >= 3.11 вы также можете использовать стандартный синтаксис except*:
# try:
#     bus.publish("data_sync")
# except* ValueError as eg:
#     print(f"Обработка ValueError из группы: {eg.exceptions}")
# except* IOError as eg:
#     print(f"Обработка IOError из группы: {eg.exceptions}")

Передача Pydantic-моделей

Если установлен пакет pydantic (chutils[pydantic]), шина поддерживает прямую передачу экземпляров Pydantic-моделей в качестве payload:

from pydantic import BaseModel
from chutils.events import subscribe, publish


class UserEvent(BaseModel):
    user_id: int
    role: str


@subscribe("user_updated")
def handle_user_update(event: UserEvent):
    print(f"Роль пользователя {event.user_id} обновлена на {event.role}")


# Публикация Pydantic-модели
publish("user_updated", UserEvent(user_id=10, role="admin"))

18. Планировщик фоновых задач (Lightweight Task Scheduler)

Модуль chutils.tasks предоставляет планировщик для выполнения периодических фоновых задач. Он поддерживает как синхронные, так и асинхронные обработчики, отслеживает перекрытия (overlapping), предоставляет стратегии обработки ошибок и автоматически интегрируется с Graceful Shutdown (chutils.lifecycle).

Простой пример

Используйте декоратор @periodic_task для регистрации задач и функцию start_scheduler() для их запуска в Event Loop.

import asyncio
import time
from chutils import periodic_task, start_scheduler, setup_graceful_shutdown
from chutils.tasks import ErrorStrategy


# 1. Асинхронная задача запускается сразу при старте планировщика
@periodic_task(interval_seconds=5, run_immediately=True, name="async_logger")
async def log_status():
    print("Статус приложения в норме... (асинхронно)")


# 2. Синхронная задача выполняется в пуле потоков каждые 10 секунд
@periodic_task(interval_seconds=10, run_immediately=False, name="sync_cleaner")
def cleanup_temp_files():
    print("Очистка временных файлов... (синхронно)")
    time.sleep(1.0)


async def main():
    # Настраиваем graceful shutdown для остановки планировщика при Ctrl+C
    setup_graceful_shutdown()

    # Запускаем планировщик фоновых задач
    start_scheduler()

    # Имитируем работу приложения
    await asyncio.sleep(30.0)


if __name__ == "__main__":
    asyncio.run(main())

Контроль перекрытия задач (Overlapping)

  • По умолчанию (overlap=False): Если предыдущий запуск задачи еще не завершился к моменту наступления следующего интервала, запуск пропускается (откладывается до следующего тика).
  • Параллельный запуск (overlap=True): Задача запускается строго по интервалу, независимо от того, завершились ли предыдущие запуски.
# Эта задача будет запускаться каждые 2 секунды параллельно, не дожидаясь окончания
@periodic_task(interval_seconds=2, overlap=True)
async def parallel_task():
    await asyncio.sleep(5.0)

Стратегии обработки ошибок (ErrorStrategy)

В случае сбоя задачи вы можете выбрать одну из трех стратегий:

  1. IGNORE (по умолчанию) — залогировать ошибку в chutils.logger и продолжить выполнение по расписанию.
  2. STOP_TASK — исключить сбойную задачу из расписания планировщика.
  3. STOP_SCHEDULER — остановить весь планировщик.
@periodic_task(interval_seconds=5, error_strategy=ErrorStrategy.STOP_TASK)
def fragile_task():
    raise RuntimeError("Неустранимая ошибка в задаче")

Динамические интервалы (Dynamic Intervals)

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

  1. Callable-функция: Передайте функцию, возвращающую int. Планировщик будет вычислять интервал заново перед каждым циклом ожидания.
  2. Конфигурация chutils.config: Передайте строку вида section.key (или просто key для поиска в секции default). Значение будет считываться из конфигурации на лету перед каждым циклом ожидания.

Важные особенности:

  • Горячая перезагрузка (Hot-Reload) "на лету": Планировщик повторно вычисляет интервал (через вызов функции или чтение конфигурации) перед каждым циклом ожидания (asyncio.sleep). Это позволяет менять расписание фоновых задач в реальном времени без необходимости перезапуска самого планировщика или всего приложения.
  • Отказоустойчивость: Если при считывании интервала возникла ошибка (например, передан неверный тип данных вроде строки вместо числа, ключ отсутствует в конфигурации, или функция выбросила исключение), планировщик логирует предупреждение (logger.warning / logger.error), автоматически переключается на безопасный резервный интервал в 1 секунду и продолжает работу. Вся служба планировщика при этом не падает.
# А. Динамический интервал через функцию
current_interval = 5


def get_interval():
    return current_interval  # Может меняться динамически во время работы


@periodic_task(interval_seconds=get_interval)
async def dynamic_job():
    print("Выполнение с динамическим интервалом...")


# Б. Динамический интервал из chutils.config (ключ 'scheduler.cleanup_delay')
@periodic_task(interval_seconds="scheduler.cleanup_delay")
async def config_bound_job():
    print("Интервал берется и динамически обновляется из конфигурации...")

19. Ограничение частоты вызовов (Rate Limiting / Throttling)

Декоратор @rate_limit из модуля chutils.decorators позволяет ограничить частоту выполнения синхронных и асинхронных функций, обеспечивая защиту от перегрузки и соблюдение лимитов внешних API.

Базовое использование (Fail Fast)

По умолчанию декоратор работает в режиме wait=False (Fail Fast) с использованием алгоритма Token Bucket. При превышении лимита мгновенно выбрасывается исключение RateLimitExceededError.

from chutils import rate_limit, RateLimitExceededError


# Разрешено максимум 3 вызова за 5 секунд
@rate_limit(max_calls=3, period=5.0)
def call_api():
    print("API вызван успешно")


for _ in range(3):
    call_api()

try:
    call_api()  # Четвертый вызов упадет
except RateLimitExceededError:
    print("Превышен лимит запросов!")

Сглаживание и ожидание (Wait)

Если передан флаг wait=True, выполнение функции блокируется на время, необходимое для восстановления лимита (для синхронных функций используется time.sleep, для асинхронных — asyncio.sleep).

import asyncio
from chutils import rate_limit


# Разрешено максимум 2 вызова в секунду. Лишние вызовы будут ожидать очереди
@rate_limit(max_calls=2, period=1.0, wait=True)
async def process_item(item_id: int):
    print(f"Обработка элемента {item_id}")


async def main():
    # Запустим 5 задач параллельно. Они будут выполняться пачками по 2 штуки в секунду
    tasks = [process_item(i) for i in range(5)]
    await asyncio.gather(*tasks)

Раздельные лимиты по ключам (key_func)

По умолчанию лимит действует глобально на всю функцию. Вы можете настроить индивидуальные лимиты (например, на каждого пользователя или IP) с помощью функции key_func.

from chutils import rate_limit


# Лимит: 1 вызов в секунду на каждого конкретного пользователя
@rate_limit(
    max_calls=1,
    period=1.0,
    key_func=lambda user_id, *args, **kwargs: f"user_{user_id}"
)
def send_notification(user_id: int, message: str):
    print(f"Уведомление отправлено пользователю {user_id}")


send_notification(1, "Привет!")  # OK
send_notification(2, "Привет!")  # OK для другого пользователя
# send_notification(1, "Еще раз привет")  # Упадет, превышен лимит для user_1

Алгоритмы лимитирования (strategy)

Поддерживаются две стратегии:

  1. token_bucket (по умолчанию) — алгоритм маркерной корзины. Разрешает кратковременные всплески нагрузки (до max_calls одновременно), после чего скорость ограничивается.
  2. leaky_bucket — алгоритм дырявого ведра. Обеспечивает строгое сглаживание нагрузки с равномерной задержкой между вызовами.
# Строгий Leaky Bucket: вызовы будут распределены равномерно
@rate_limit(max_calls=60, period=60.0, strategy="leaky_bucket", wait=True)
def send_email():
    pass

20. Внедрение зависимостей (Dependency Injection)

Встроенный IoC/DI контейнер (chutils.di) позволяет связать независимые компоненты приложения без ручной передачи аргументов через всю цепочку вызовов (prop drilling) и без привязки к жестким глобальным зависимостям.

Декларативная регистрация (@provide)

Используйте декоратор @provide для регистрации классов или функций-фабрик. По умолчанию используется время жизни singleton (объект создается один раз при первом обращении и кэшируется).

from chutils import provide


# 1. Регистрация класса (singleton)
@provide()
class DatabaseConnection:
    def __init__(self) -> None:
        self.connected = True
        print("Подключение к БД создано")


# 2. Регистрация функции-фабрики (transient - новый объект при каждом запросе)
@provide(scope="transient")
def get_current_time() -> float:
    import time
    return time.time()

Автоматическое внедрение (@inject)

Декоратор @inject автоматически подставляет зарегистрированные зависимости из контейнера в аргументы функции. Для этого аргумент должен иметь тип зарегистрированной зависимости и значение по умолчанию Inject().

from chutils import inject, Inject


# Автоматически внедряем DatabaseConnection
@inject()
def get_user_profile(user_id: int, db: DatabaseConnection = Inject()):
    if db.connected:
        return f"User profile {user_id}"

Рекурсивный резолв зависимостей

Если одна зависимость требует другую зависимость в своем конструкторе __init__, контейнер автоматически разрешит весь граф при обращении (Auto-wiring).

from chutils import provide, inject, Inject


@provide()
class CacheService:
    pass


@provide()
class UserService:
    # Контейнер автоматически создаст CacheService и передаст его сюда
    def __init__(self, cache: CacheService) -> None:
        self.cache = cache


@inject()
def process(user_service: UserService = Inject()):
    # user_service уже имеет внутри инициализированный cache
    pass

Тестирование и переопределение зависимостей

Для написания unit-тестов вы можете легко переопределить любую зависимость в контейнере (mocking) напрямую через метод register() глобального контейнера:

from chutils import container


class MockDatabaseConnection:
    def __init__(self) -> None:
        self.connected = True
        print("Mock подключение к БД создано!")


def test_my_service():
    # Переопределяем реальное подключение на мок
    container.register(DatabaseConnection, provider=MockDatabaseConnection)

    # Теперь все вызовы @inject внедрят MockDatabaseConnection вместо реального
    profile = get_user_profile(42)
    assert "User profile 42" in profile

21. Сбор метрик (Unified Metrics)

Модуль chutils.metrics предоставляет унифицированный API для сбора многомерных метрик (Counters, Gauges, Histograms) с полной поддержкой меток (labels).

Быстрый старт (Counters & Gauges)

По умолчанию метрики сохраняются во встроенное in-memory хранилище. При установке библиотеки prometheus-client они автоматически транслируются в формат Prometheus.

from chutils.metrics import increment, set_gauge, observe

# 1. Увеличить счетчик (Counter)
increment("http_requests_total", 1.0, {"method": "GET", "endpoint": "/users"})

# 2. Установить значение датчика (Gauge)
set_gauge("cpu_usage_percent", 45.2, {"node": "worker-1"})

# 3. Записать значение в гистограмму (Histogram)
observe("request_size_bytes", 1024.0, {"client": "ios"})

Автоматические метрики Circuit Breaker (Предохранителя)

Декоратор @circuit_breaker автоматически экспортирует метрику состояния цепи типа Gauge под именем circuit_breaker_state с меткой name (имя декорируемой функции).

Метрика состояния кодируется числовыми значениями:

  • 0.0 — CLOSED (цепь замкнута, всё работает нормально).
  • 1.0 — OPEN (цепь разомкнута из-за сбоев, вызовы заблокированы).
  • 2.0 — HALF_OPEN (период восстановления, пробные вызовы).

Это позволяет визуализировать аварийные режимы и работоспособность интеграций в реальном времени (например, на дашборде Grafana).

Автоматический замер времени (@timer)

Используйте @timer в качестве декоратора (для синхронных и асинхронных функций) или в качестве контекстного менеджера.

from chutils.metrics import timer
import asyncio


# A. Декоратор (асинхронный)
@timer("http_request_duration_seconds", labels={"handler": "get_users"})
async def handle_request():
    await asyncio.sleep(0.05)


# B. Контекстный менеджер
def run_db_query():
    with timer("db_query_duration_seconds", labels={"query": "select_orders"}):
        # тяжелая операция
        pass

Экспорт метрик в FastAPI / Flask

Чтобы отдавать собранные метрики в систему мониторинга Prometheus, просто добавьте эндпоинт /metrics, вызывающий generate_latest():

from fastapi import FastAPI, Response
from chutils.metrics import generate_latest

app = FastAPI()


@app.get("/metrics")
def metrics_endpoint():
    # generate_latest возвращает валидный Prometheus-текст
    return Response(content=generate_latest(), media_type="text/plain")

Безопасность при отсутствии prometheus-client (Graceful Fallback)

Модуль chutils.metrics спроектирован так, что отсутствие внешней библиотеки prometheus-client не вызывает ошибок. В этом случае все метрики собираются во внутренний in-memory пул и форматируются функцией generate_latest() в полностью Prometheus-совместимый текстовый формат.

Таким образом, ваше приложение гарантированно продолжит работу без внешних зависимостей.

22. Универсальная валидация данных (chutils.validation)

Модуль chutils.validation предоставляет удобный инструмент для валидации структурированных данных (JSON, dict) с использованием Pydantic, а также автоматическую валидацию аргументов функций с генерацией детальных, структурированных ошибок ChutilsValidationError.

Валидация словарей и JSON-строк (validate_data)

Функция validate_data принимает Pydantic модель и данные (словарь или JSON-строку), возвращая валидный объект модели. В случае ошибки выбрасывается ChutilsValidationError.

from pydantic import BaseModel, Field
from chutils import validate_data
from chutils.exceptions import ChutilsValidationError


class User(BaseModel):
    name: str
    age: int = Field(gt=0)


try:
    # 1. Валидация словаря
    user1 = validate_data(User, {"name": "Alice", "age": 30})

    # 2. Валидация JSON-строки
    user2 = validate_data(User, '{"name": "Bob", "age": 25}')
except ChutilsValidationError as e:
    print(f"Ошибка: {e}")

Автоматическая валидация аргументов функций (@validate_call)

Декоратор @validate_call автоматически проверяет типы и значения аргументов при вызове функции. Поддерживает как синхронные, так и асинхронные функции.

from chutils import validate_call
from chutils.exceptions import ChutilsValidationError


@validate_call
def send_message(user_id: int, message: str) -> None:
    print(f"Отправка пользователю {user_id}: {message}")


try:
    # Вызовет ошибку валидации, так как передан неверный тип
    send_message("not_an_int", "Привет")
except ChutilsValidationError as e:
    print(f"Неверные аргументы вызова:\n{e}")

Красивый вывод ошибок через Rich

Исключение ChutilsValidationError бесшовно интегрировано с библиотекой rich. Если rich установлен, при печати исключения в консоль автоматически выводится красивая структурированная таблица ошибок с указанием пути к полю, причины ошибки и полученного невалидного значения.

from chutils import get_console
from chutils.exceptions import ChutilsValidationError

console = get_console()

try:
    validate_data(User, {"name": "Alice", "age": -5})
except ChutilsValidationError as e:
    # Автоматически отрендерит красивую таблицу ошибок в консоль
    console.print(e)

23. Мониторинг работоспособности и Diagnostics API (chutils.diagnostics)

Модуль chutils.diagnostics предназначен для проверки жизнедеятельности (health check) компонентов системы с контролем таймаутов, встроенными проверками и готовыми хелперами для веб-фреймворков.

Использование DiagnosticsManager

Вы можете зарегистрировать кастомные функции проверок (как синхронные, так и асинхронные) с указанием таймаутов выполнения.

import asyncio
from chutils.diagnostics import DiagnosticsManager

manager = DiagnosticsManager()


# Регистрация асинхронной проверки с таймаутом 2 секунды
@manager.register(name="database", critical=True, timeout=2.0)
async def check_db() -> str:
    await asyncio.sleep(0.1)  # Симуляция пинга БД
    return "Соединение с БД успешно установлено."


# Запуск проверок асинхронно
async def main():
    report = await manager.run_checks()
    print(f"Статус системы: {report.status}")  # HEALTHY, DEGRADED, UNHEALTHY
    for name, result in report.results.items():
        print(f"[{name}] {result.status} - {result.message}")


asyncio.run(main())

Встроенные проверки

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

  • check_keyring — проверяет доступность системного хранилища секретов (keyring) путём пробной записи и удаления тестового секрета.
  • check_config — проверяет валидность и физическое наличие конфигурационного файла на диске.

Встроенные проверки зарегистрированы в DiagnosticsManager по умолчанию.

Интеграция с FastAPI и Flask

Модуль содержит готовые хелперы для быстрой отдачи статуса здоровья системы по HTTP-эндпоинтам.

FastAPI:

from fastapi import FastAPI
from chutils.diagnostics import DiagnosticsManager
from chutils.diagnostics.web import get_fastapi_health_handler

app = FastAPI()
manager = DiagnosticsManager()

# Регистрируем роут /health
app.add_api_route("/health", get_fastapi_health_handler(manager), methods=["GET"])

При возникновении критических сбоев (UNHEALTHY) обработчик автоматически возвращает HTTP-код 503 Service Unavailable, а при успешной или частично деградировавшей работе (HEALTHY, DEGRADED) — 200 OK.

Проверка через CLI

Вы можете запустить диагностику системы напрямую из терминала:

# Красивая таблица результатов в rich
chutils dev diagnostics

# Вывод в формате структурированного JSON для систем мониторинга
chutils dev diagnostics --json

24. Декларативный манифест переменных окружения (chutils.env)

Модуль chutils.env предоставляет механизм для декларативного описания, загрузки и валидации переменных окружения с использованием Pydantic моделей, поддержкой автоматического маскирования секретов и интеграцией с SecretManager.

Создание манифеста

Опишите ожидаемые переменные окружения как поля класса, наследующегося от BaseEnvManifest:

from pydantic import Field

from chutils.env import BaseEnvManifest


class AppEnv(BaseEnvManifest):
    DATABASE_URL: str = Field(description="URL подключения к базе данных")
    API_KEY: str = Field(json_schema_extra={"secret": True}, description="Секретный API-ключ")
    PORT: int = Field(default=8080, description="Порт веб-сервера")

Загрузка и валидация в коде

Метод load() автоматически считывает значения из os.environ и выполняет приведение типов:

import os
from chutils.exceptions import EnvValidationError

# Устанавливаем переменные для теста
os.environ["DATABASE_URL"] = "postgresql://localhost:5432/db"
os.environ["API_KEY"] = "super-secret-token"

try:
    env = AppEnv.load()
    print(f"Server will run on port {env.PORT}")
except EnvValidationError as e:
    # Все секретные поля (API_KEY) будут автоматически замаскированы (показано как ***)
    print(f"Ошибка валидации окружения: {e}")

Если переменная отсутствует в os.environ, но помечена как secret=True, метод load() автоматически попытается получить её значение через SecretManager (если он доступен).

Валидация окружения через CLI

Вы можете автоматически валидировать переменные окружения при развертывании приложения (например, в Docker/K8s init-контейнерах) с помощью CLI-команды chutils env validate.

  1. Явный запуск с указанием манифеста:
chutils env validate --manifest myapp.env:AppEnv
  1. Запуск через конфигурацию проекта (pyproject.toml):

Добавьте секцию в ваш pyproject.toml:

[tool.chutils.env]
manifest = "myapp.env:AppEnv"

После этого вы можете запускать проверку без параметров:

chutils env validate

При успешной валидации команда вернет код 0, при сбое — выведет подробную таблицу ошибок и вернет код 1.