Справочник API

В этом разделе находится документация, автоматически сгенерированная из исходного кода chutils. Все детали реализации, приоритеты настроек и примеры перенесены непосредственно в докстринги модулей и функций.

Пакет chutils

chutils

Пакет chutils - набор переиспользуемых утилит для Python.

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

Ключевые особенности: - Автоматическое обнаружение корня проекта и файла конфигурации. - Поддержка форматов config.yml, config.yaml и config.ini (YAML в приоритете). - Удобные функции для доступа к настройкам, включая разрешение путей. - Асинхронные версии основных функций для неблокирующей работы. - Готовый к работе логгер с выводом в консоль и ротируемые файлы. - Безопасное хранение секретов через системное хранилище (keyring).

Основное использование:

Вам не нужно ничего инициализировать. Просто импортируйте и используйте:

from chutils import get_config_value, setup_logger, SecretManager

logger = setup_logger()
secrets = SecretManager("my_app")
db_host = get_config_value("Database", "host", "localhost")
logger.info(f"Подключение к базе данных на {db_host}")

Ручная инициализация (для нестандартных случаев):

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

import chutils
chutils.init(base_dir="/path/to/your/project")

__dir__()

Возвращает список всех доступных атрибутов для поддержки автодополнения и интроспекции.

__getattr__(name)

Реализация ленивой загрузки согласно PEP 562. Вызывается при обращении к атрибутам модуля, которые не определены явно.

init(base_dir)

Ручная инициализация пакета с указанием базовой директории проекта.

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

Parameters:

Name	Type	Description	Default
base_dir	str	Абсолютный путь к корневой директории проекта.	required

Raises:

Type	Description
ChutilsException	Если указанная директория не существует.

options: members: [init]

Модуль config

chutils.config

Модуль для работы с конфигурацией.

Обеспечивает автоматический поиск файла config.yml, config.yaml или config.ini в корне проекта и предоставляет удобные функции для чтения и сохранения настроек. Поддерживает кастомные уровни логирования при условии, что модуль logger загружен.

Переопределение конфигурации

Библиотека поддерживает многоуровневое переопределение настроек: 1. Переменные окружения (CH_[SECTION]_[KEY]): Имеют наивысший приоритет. 2. Локальный файл (config.local.yml): Переопределяет значения основного файла. 3. Основной файл (config.yml): Базовые настройки проекта.

Локальные файлы конфигурации (например, config.local.yml или config.local.ini) должны находиться в той же директории, что и основной файл. Это позволяет удобно управлять чувствительными или специфичными для разработчика настройками, не коммитя их в репозиторий.

__getattr__(name)

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

Согласно PEP 562, эта функция вызывается при обращении к отсутствующим атрибутам модуля. Мы используем её для перенаправления обращений к старым приватным переменным в новый менеджер состояния с выводом предупреждения об устаревании.

aget_config(model=None) async

Асинхронная версия get_config.

Parameters:

Name	Type	Description	Default
model	Optional[Type[T]]	Опциональный класс Pydantic модели для валидации.	None

Returns:

Type	Description
Union[Dict[str, Any], T]	Словарь конфигурации или экземпляр Pydantic модели.

are_paths_initialized()

Проверяет, были ли инициализированы пути к проекту и файлам конфигурации.

Returns:

Type	Description
bool	True, если пути определены.

asave_config_value(section, key, value, cfg_file=None, save_to_local=False, notify=True) async

Асинхронно сохраняет одно значение в конфигурационном файле. Работает как асинхронная обертка вокруг синхронной save_config_value().

Parameters:

Name	Type	Description	Default
section	str	Имя секции.	required
key	str	Имя ключа в секции.	required
value	Any	Новое значение для ключа.	required
cfg_file	Optional[str]	Опциональный путь к файлу для сохранения. Если указан, имеет приоритет над save_to_local.	None
save_to_local	bool	Если True, и существует локальный файл конфигурации (например, config.local.yml), значение будет сохранено в него. По умолчанию False.	False
notify	bool	Если True (по умолчанию), Hot-Reload watcher уведомит о смене конфигурации. Если False, уведомление будет подавлено.	True

Returns:

Name	Type	Description
True	bool	Если значение было успешно обновлено и сохранено.
False	bool	Если файл не найден, или произошла ошибка.

export_schema(model, output_path=None, indent=4)

Генерирует JSON Schema для Pydantic модели и опционально сохраняет в файл.

Parameters:

Name	Type	Description	Default
model	Union[Type[BaseModel], str]	Класс модели или строковый путь к нему ('module:Class').	required
output_path	Optional[Union[str, Path]]	Путь к файлу для сохранения схемы.	None
indent	int	Отступ в результирующем JSON.	4

Returns:

Type	Description
str	Строка с JSON Schema.

generate_env_template(model_class, prefix='CH')

Генерирует .env шаблон (плоский формат).

generate_json_schema(model_class)

Генерирует JSON схему на основе Pydantic модели.

generate_yaml_template(model_class, indent=0)

Генерирует YAML шаблон на основе Pydantic модели с комментариями.

get_all_config_paths(cfg_file=None)

Возвращает пути к основному, специфичному для окружения и локальному файлам конфигурации.

Parameters:

Name	Type	Description	Default
cfg_file	Optional[str]	Опциональный путь к основному файлу.	None

Returns:

Type	Description
Tuple[Optional[str], Optional[str], Optional[str]]	Кортеж (путь_к_основному, путь_к_окружению, путь_к_локальному).

get_base_dir()

Возвращает абсолютный путь к корневой директории проекта.

Если пути еще не инициализированы, запускает автоматический поиск.

Returns:

Type	Description
Optional[str]	Путь к корню проекта или None, если корень не найден.

get_config(model=None, remote_url=None, remote_auth=None, polling_interval=None)

Загружает и объединяет конфигурацию из всех доступных источников.

Результат кэшируется. Повторные вызовы возвращают кэшированный объект, если он не был сброшен (например, при сохранении нового значения).

Порядок применения конфигураций (от меньшего приоритета к большему): 1. Основной файл (config.yml) 2. Файл окружения (config.{CH_ENV}.yml) 3. Локальный файл (config.local.yml) 4. Удаленный источник (если указан remote_url) 5. Переменные окружения (CH_SECTION_KEY)

Parameters:

Name	Type	Description	Default
model	Optional[Type[T]]	Опциональный класс Pydantic модели для валидации.	None
remote_url	Optional[str]	URL для загрузки удаленной конфигурации.	None
remote_auth	Optional[Tuple[str, str]]	Кортеж (login, password) для Basic Auth.	None
polling_interval	Optional[int]	Интервал опроса удаленного источника в секундах. Если не указан, опрос не запускается.	None

Returns:

Type	Description
Union[Dict[str, Any], T]	Словарь со всей конфигурацией проекта или экземпляр Pydantic модели.
Union[Dict[str, Any], T]	Если файлы не найдены, возвращается пустой словарь (или ошибка валидации модели).

Raises:

Type	Description
ConfigLoadError	Если произошла ошибка при чтении файлов конфигурации.
ConfigParseError	Если файлы конфигурации содержат синтаксические ошибки.
OptionalDependencyError	Если передана model, но пакет pydantic не установлен.

get_config_boolean(section, key, fallback=False, config=None)

Получает булево значение из конфигурации.

Распознает 'true', '1', 't', 'y', 'yes' как True и 'false', '0', 'f', 'n', 'no' как False (без учета регистра).

Parameters:

Name	Type	Description	Default
section	str	Имя секции.	required
key	str	Имя ключа.	required
fallback	bool	Значение по умолчанию, если ключ не найден или не может быть распознан как булево.	False
config	Optional[Dict]	Опциональный, предварительно загруженный словарь конфигурации.	None

Returns:

Type	Description
bool	True или False.

get_config_file_path()

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

Returns:

Type	Description
Optional[str]	Путь к файлу или None, если файл не найден.

get_config_float(section, key, fallback=0.0, config=None)

Получает дробное значение из конфигурации.

Parameters:

Name	Type	Description	Default
section	str	Имя секции.	required
key	str	Имя ключа.	required
fallback	float	Значение по умолчанию, если ключ не найден или не может быть преобразован в float.	0.0
config	Optional[Dict]	Опциональный, предварительно загруженный словарь конфигурации.	None

Returns:

Type	Description
float	Float или fallback.

get_config_int(section, key, fallback=0, config=None)

Получает целочисленное значение из конфигурации.

Parameters:

Name	Type	Description	Default
section	str	Имя секции.	required
key	str	Имя ключа.	required
fallback	int	Значение по умолчанию, если ключ не найден или не может быть преобразован в int.	0
config	Optional[Dict]	Опциональный, предварительно загруженный словарь конфигурации.	None

Returns:

Type	Description
int	Целое число из конфигурации или fallback.

get_config_list(section, key, fallback=None, config=None)

Получает значение как список из конфигурации.

Parameters:

Name	Type	Description	Default
section	str	Имя секции.	required
key	str	Имя ключа.	required
fallback	Optional[List[Any]]	Значение по умолчанию, если ключ не найден.	None
config	Optional[Dict]	Опциональный, предварительно загруженный словарь конфигурации.	None

Returns:

Type	Description
List[Any]	Список из конфигурации или fallback. Если fallback не указан,
List[Any]	возвращается пустой список.

get_config_path(section, key, fallback=None, config=None, resolve_from_root=True)

Получает путь из конфигурации. Функция автоматически добавляет _BASE_DIR к относительным путям, если resolve_from_root установлено в True. Args: section: Имя секции. key: Имя ключа. fallback: Значение по умолчанию, если ключ не найден. config: Опциональный, предварительно загруженный словарь конфигурации. resolve_from_root: Если True, относительные пути будут разрешаться относительно _BASE_DIR. Если False, пути возвращаются как есть, без добавления _BASE_DIR. Returns: Путь из конфигурации или fallback.

get_config_paths(cfg_file=None)

Возвращает пути к основному и локальному файлам конфигурации.

Legacy API для обратной совместимости. Возвращает кортеж из 2 элементов. Для получения всех путей (включая env) используйте get_all_config_paths().

Parameters:

Name	Type	Description	Default
cfg_file	Optional[str]	Опциональный путь к основному файлу.	None

Returns:

Type	Description
Tuple[Optional[str], Optional[str]]	Кортеж (путь_к_основному, путь_к_локальному).

get_config_section(section_name, fallback=None, config=None, model=None)

get_config_section(
    section_name: str,
    fallback: Optional[Dict] = None,
    config: Optional[Dict] = None,
    model: None = None,
) -> Dict[str, Any]

get_config_section(
    section_name: str,
    fallback: Optional[Dict] = None,
    config: Optional[Dict] = None,
    model: Type[T] = None,
) -> T

Получает всю секцию конфигурации как словарь или Pydantic модель.

Parameters:

Name	Type	Description	Default
section_name	str	Имя секции.	required
fallback	Optional[Dict]	Значение по умолчанию, если секция не найдена.	None
config	Optional[Dict]	Опциональный, предварительно загруженный словарь конфигурации.	None
model	Optional[Type[T]]	Опциональный класс Pydantic модели для валидации секции.	None

Returns:

Type	Description
Union[Dict[str, Any], T]	Словарь с содержимым секции или экземпляр Pydantic модели.
Union[Dict[str, Any], T]	Если fallback не указан и секция не найдена, возвращается пустой словарь.

Raises:

Type	Description
ConfigLoadError	Если произошла ошибка при чтении файлов конфигурации.
ConfigParseError	Если файлы конфигурации содержат синтаксические ошибки.
OptionalDependencyError	Если передана model, но пакет pydantic не установлен.

get_config_value(section, key, fallback=None, config=None)

Получает произвольное значение из конфигурации.

Если значение не найдено или оно пустое, возвращает fallback. Поддерживает универсальное переопределение через переменные окружения по шаблону CH_[SECTION]_[KEY], если не установлено CH_DISABLE_ENV_OVERRIDE=true.

Parameters:

Name	Type	Description	Default
section	str	Имя секции.	required
key	str	Имя ключа.	required
fallback	Any	Значение по умолчанию, если ключ не найден или его значение пустое.	None
config	Optional[Dict]	Опциональный, предварительно загруженный словарь конфигурации.	None

Returns:

Type	Description
Any	Значение из конфигурации или fallback.

import_model_class(model_path)

Импортирует класс Pydantic модели по строковому пути.

Parameters:

Name	Type	Description	Default
model_path	str	Путь к модели в формате 'module.path:ClassName'.	required

Returns:

Type	Description
Type[BaseModel]	Класс модели (BaseModel).

Raises:

Type	Description
ValueError	Если формат пути некорректен.
ImportError	Если модуль или класс не найдены.
TypeError	Если найденный объект не является подклассом BaseModel.

is_config_loaded()

Проверяет, была ли конфигурация уже загружена в память.

Returns:

Type	Description
bool	True, если кэш конфигурации заполнен.

on_config_change(callback)

Регистрирует функцию обратного вызова, которая будет вызвана при изменении конфигурации.

Parameters:

Name	Type	Description	Default
callback	Callable[[], None]	Функция без аргументов.	required

save_config_value(section, key, value, cfg_file=None, save_to_local=False, notify=True)

Сохраняет или обновляет одно значение в файле конфигурации.

Warning

Важно: При сохранении в .yml комментарии и форматирование будут утеряны. При сохранении в .ini - сохраняются.

Parameters:

Name	Type	Description	Default
section	str	Имя секции.	required
key	str	Имя ключа в секции.	required
value	Any	Новое значение для ключа.	required
cfg_file	Optional[str]	Опциональный путь к файлу для сохранения. Если указан, имеет приоритет над save_to_local.	None
save_to_local	bool	Если True, и существует локальный файл конфигурации (например, config.local.yml), значение будет сохранено в него. По умолчанию False.	False
notify	bool	Если True (по умолчанию), Hot-Reload watcher уведомит о смене конфигурации. Если False, уведомление будет подавлено.	True

Returns:

Name	Type	Description
True	bool	Если значение было успешно обновлено и сохранено.
False	bool	Если файл не найден, или произошла ошибка.

start_config_watcher()

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

Требует установленного пакета watchdog.

Returns:

Type	Description
bool	True, если watcher успешно запущен.

Raises:

Type	Description
OptionalDependencyError	Если пакет watchdog не установлен.

stop_config_watcher()

Останавливает процесс отслеживания изменений конфигурации.

options: members:

• get_config
• aget_config
• get_config_value
• HttpConfigProvider
• get_config_int
• get_config_float
• get_config_boolean
• get_config_list
• get_config_section
• get_config_path
• save_config_value
• asave_config_value
• start_config_watcher
• stop_config_watcher
• on_config_change
• get_base_dir
• get_config_file_path
• is_config_loaded
• are_paths_initialized
• get_config_paths
• generate_yaml_template
• generate_env_template
• generate_json_schema
• export_schema
• import_model_class

Модуль logger

chutils.logger

Модуль для настройки логирования.

Этот пакет разделен на модули для соблюдения SRP: - core: Основной класс логгера и setup_logger. - masking: Фильтрация секретов. - formatters: Форматирование (Text, JSON). - handlers: Обработчики файлов (ротация, сжатие).

ChutilsJsonFormatter

Bases: JsonFormatter

Кастомный JSON-форматтер, который группирует контекстные данные во вложенный объект 'context', а данные трассировки выносит на верхний уровень.

ChutilsLogger

Bases: Logger

Кастомный класс логгера, расширяющий стандартный logging.Logger.

Добавляет поддержку пользовательских уровней логирования (devdebug и mediumdebug), обеспечивая при этом корректную работу статических анализаторов и автодополнения в IDE.

Иерархия уровней

• DEVDEBUG (9): Максимально подробный вывод для глубокой отладки. Предназначен для вывода дампов переменных, внутренних состояний и т.д.
• DEBUG (10): Стандартный отладочный уровень.
• MEDIUMDEBUG (15): Промежуточный уровень между DEBUG и INFO. Полезен для менее критичной, но более подробной, чем INFO, информации.
• INFO (20): Стандартный информационный уровень.

Note

Не создавайте экземпляр напрямую. Используйте setup_logger().

Example

logger: ChutilsLogger = setup_logger()
logger.devdebug("Максимально подробное сообщение")

add_mask(value)

Добавляет строку в глобальный список маскируемых секретов.

Каждая зарегистрированная строка будет заменяться на '***' во всех сообщениях всех логгеров chutils.

Parameters:

Name	Type	Description	Default
value	str	Секретная строка для маскирования.	required

devdebug(message, *args, **kws)

Логирует сообщение с уровнем DEVDEBUG (9).

Parameters:

Name	Type	Description	Default
message	str	Сообщение для логирования.	required
*args	Any	Аргументы форматирования.	()
**kws	Any	Ключевые слова для логгера.	{}

mediumdebug(message, *args, **kws)

Логирует сообщение с уровнем MEDIUMDEBUG (15).

Parameters:

Name	Type	Description	Default
message	str	Сообщение для логирования.	required
*args	Any	Аргументы форматирования.	()
**kws	Any	Ключевые слова для логгера.	{}

CompressingRotatingFileHandler

Bases: RotatingFileHandler

Обработчик ротации по размеру с поддержкой сжатия (gzip).

Обеспечивает корректную работу с цепочкой сжатых бэкапов.

doRollover()

Выполняет ротацию логов с последующим сжатием старого файла.

Процесс: 1. Закрытие текущего потока. 2. Сдвиг существующих архивов (log.1.gz -> log.2.gz). 3. Переименование текущего лога в log.1. 4. Открытие нового файла для дальнейшей записи. 5. Сжатие переименованного файла в фоне.

CompressingTimedRotatingFileHandler

Bases: SafeTimedRotatingFileHandler

Обработчик ротации по времени с поддержкой сжатия (gzip).

doRollover()

Выполняет временную ротацию и сжимает полученные бэкапы.

LogLevel

Bases: str, Enum

Перечисление для поддерживаемых уровней логирования.

SafeTimedRotatingFileHandler

Bases: TimedRotatingFileHandler

Надежный обработчик ротации логов, адаптированный для Windows.

Этот класс решает проблему PermissionError при ротации логов в Windows, гарантируя, что файл будет закрыт перед переименованием. Он явно закрывает файловый поток перед вызовом стандартной логики ротации.

doRollover()

Выполняет ротацию, закрывая текущий поток перед операцией.

SecretMaskingFilter

Bases: Filter

Фильтр для автоматического маскирования секретов в сообщениях логов.

Ищет в тексте сообщения и в аргументах все зарегистрированные секреты и паттерны и заменяет их на '[MASKED]'.

filter(record)

Применяет маскирование к записи лога.

Parameters:

Name	Type	Description	Default
record	LogRecord	Запись лога.	required

Returns:

Type	Description
bool	Всегда True (фильтр не отсеивает записи, а модифицирует их).

setup_logger(name='app_logger', config_section_name=None, log_level=None, log_file_name=None, force_reconfigure=False, rotation_type=None, max_bytes=None, compress=None, backup_count=None, encoding=None, when=None, interval=None, utc=None, at_time=None, json_format=None, use_async=None, custom_patterns=None, use_predefined_patterns=None, **kwargs)

Настраивает и возвращает экземпляр логгера.

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

Приоритет настроек:

1. Переменные окружения CH_LOG_NO_TIME и CH_LOG_NO_FILE (высший приоритет).
2. Явные аргументы, переданные в эту функцию (например, log_level=...).
3. Секция, указанная в config_section_name (например, [AuditLogger]).
4. Общая секция [Logging] в config.yml.
5. Значения по умолчанию, зашитые в коде.

Ротация и сжатие:

• По времени (rotation_type='time'): Ротация ежедневно или по интервалу (параметры when, interval).
• По размеру (rotation_type='size'): Ротация при достижении max_bytes.
• Сжатие: Если compress=True, старые логи сжимаются в .gz.

Асинхронность:

• Если use_async=True, логи записываются в очередь и обрабатываются в отдельном потоке.
• Очередь блокирующая, что предотвращает потерю сообщений при переполнении.

Маскирование:

• Автоматическая замена секретов и паттернов на [MASKED].
• Поддержка предустановленных паттернов PII (email, phone, credit_card, ssn).

Parameters:

Name	Type	Description	Default
name	str	Имя логгера. app_logger по умолчанию.	'app_logger'
config_section_name	Optional[str]	Имя секции в конфиге (например, 'MyAuditLogger'). Если указана, настройки из этой секции переопределяют настройки из общей секции [Logging]. Если не указана, используется только общая секция [Logging].	None
log_level	Optional[LogLevel]	Уровень логирования (строка или LogLevel).	None
log_file_name	Optional[str]	Имя файла лога. Если не указано, берется из конфига или 'app.log'.	None
force_reconfigure	bool	Если True, пересоздает обработчики (обычно они идемпотентны).	False
rotation_type	Optional[str]	Тип ротации ('time' или 'size').	None
max_bytes	Optional[int]	Макс. размер файла (для 'size'). По умолчанию 5 МБ.	None
compress	Optional[bool]	Сжимать ли старые логи в .gz. По умолчанию False.	None
backup_count	Optional[int]	Количество хранимых ротированных файлов. По умолчанию 3.	None
encoding	Optional[str]	Кодировка файла. По умолчанию 'utf-8'.	None
when	Optional[str]	Интервал ротации для 'time' (например, 'S', 'M', 'H', 'D', 'midnight', 'W0'-'W6').	None
interval	Optional[int]	Для 'time'. Кратность интервала.	None
utc	Optional[bool]	Для 'time'. Использовать ли UTC время для имен файлов.	None
at_time	Optional[time]	Для 'time'. Время ротации (для when='midnight').	None
json_format	Optional[bool]	Использовать ли JSON формат для логов.	None
use_async	Optional[bool]	Использовать ли асинхронное логирование.	None
custom_patterns	Optional[List[str]]	Список регулярных выражений для маскирования.	None
use_predefined_patterns	Optional[List[Union[str, List[str]]]]	Список имен предустановленных паттернов для маскирования.	None
**kwargs	Any	Дополнительные параметры для FileHandler (например, delay=True, errors='ignore', mode='a').	{}

Returns:

Type	Description
ChutilsLogger	Настроенный экземпляр ChutilsLogger.

options: members:

• setup_logger
• ChutilsLogger
• DEVDEBUG_LEVEL_NUM
• MEDIUMDEBUG_LEVEL_NUM

Модуль context

chutils.context

ContextFilter

Bases: Filter

Фильтр, обогащающий LogRecord данными из контекста.

Добавляет: - Индивидуальные ключи контекста как атрибуты (для %(key)s). - record.context: Строка вида "[key1=val1 key2=val2 ]" или "" если пусто. - record.context_dict: Оригинальный словарь контекста (для JSON-логирования).

bind_context(**kwargs)

Привязывает значения к текущему контексту. Возвращает токен для последующей очистки через unbind_context.

clear_context()

Полностью очищает текущий контекст.

get_context()

Возвращает копию текущего контекста.

unbind_context(token)

Восстанавливает контекст до состояния, предшествующего bind_context.

options: members:

• bind_context
• unbind_context
• clear_context
• ContextFilter

Модуль lifecycle (Управление жизненным циклом)

chutils.lifecycle

Управление жизненным циклом приложения.

Обеспечивает механизмы регистрации функций очистки (cleanup callbacks), которые будут выполнены при завершении работы приложения.

LifecycleManager

Менеджер жизненного цикла, управляющий реестром функций очистки.

get_cleanup_callbacks()

Возвращает список зарегистрированных функций в порядке LIFO.

register_cleanup(func)

Регистрирует функцию для выполнения при завершении работы.

Функции выполняются в порядке LIFO (Last-In-First-Out). Поддерживаются как синхронные, так и асинхронные функции.

Parameters:

Name	Type	Description	Default
func	CleanupCallback	Функция или корутина для регистрации.	required

Returns:

Type	Description
CleanupCallback	Та же функция (позволяет использовать как декоратор).

setup_graceful_shutdown(signals=None)

Настраивает перехват сигналов завершения работы.

register_cleanup(func)

Регистрирует функцию очистки в менеджере.

Эта функция является публичным API для добавления колбэков, которые будут вызваны при завершении работы приложения.

Parameters:

Name	Type	Description	Default
func	CleanupCallback	Функция-колбэк, которую нужно зарегистрировать. Должна соответствовать типу CleanupCallback.	required

Returns:

Type	Description
CleanupCallback	Зарегистрированная функция (возвращает тот же объект для
CleanupCallback	использования в качестве декоратора).

Example

Использование в качестве декоратора: @register_cleanup async def close_db(): await db.close()

Использование как обычной функции: def cleanup_logs(): print("Cleaning up logs...") register_cleanup(cleanup_logs)

setup_graceful_shutdown()

Публичный API для настройки Graceful Shutdown.

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

options: members:

• register_cleanup
• setup_graceful_shutdown

Модуль cli_booster (Быстрое создание CLI)

chutils.cli_booster

Модуль для быстрого создания консольных команд (CLI Booster).

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

cli_command(func)

Декоратор для превращения функции в CLI-команду.

Интроспектирует сигнатуру функции и создает парсер аргументов argparse. Поддерживает аннотации типов, значения по умолчанию, асинхронные функции и автоматический парсинг докстрингов (Google-style) для генерации справки.

Example

@cli_command def my_script(name: str, count: int = 1): """

Пример скрипта.

Args:
    name (str): Имя пользователя.
    count (int): Количество повторений.
"""

for _ in range(count):
    print(f"Hello, {name}!")

options: members:

• cli_command

Модуль time (Работа со временем)

chutils.time

Модуль для работы со временем и датами. Обеспечивает UTC-first подход, корректную обработку часовых поясов и парсинг.

humanize_timedelta(dt, locale='ru', custom_locales=None)

Превращает дату в человекочитаемую строку относительно текущего времени.

Parameters:

Name	Type	Description	Default
dt	datetime	Дата для сравнения.	required
locale	str	Код локали ('ru' или 'en').	'ru'
custom_locales	Optional[dict]	Дополнительные локали или переопределения.	None

Returns:

Type	Description
str	Строка вида "5 минут назад", "вчера" и т.д.

parse_datetime(value)

Парсит дату и время из различных форматов и приводит к UTC aware объекту.

Поддерживаемые форматы: - ISO 8601 строки (например, "2023-01-01T12:00:00"). - UNIX timestamps в секундах (int/float). - UNIX timestamps в миллисекундах (длинные числа).

Если установлена библиотека python-dateutil (chutils[date]), поддерживается более широкий спектр форматов.

Parameters:

Name	Type	Description	Default
value	Union[str, int, float]	Строка с датой или числовое представление (timestamp).	required

Returns:

Type	Description
datetime	Объект datetime в зоне UTC.

Raises:

Type	Description
ValueError	Если формат не распознан.

utc_now()

Возвращает текущее время в формате UTC с информацией о часовом поясе.

Returns:

Type	Description
datetime	Объект datetime, представляющий текущее время в UTC.

options: members:

• utc_now
• parse_datetime
• humanize_timedelta

Модуль tracing (Распределенное трассирование)

chutils.tracing

Интеграция с OpenTelemetry для распределенного трассирования.

Позволяет автоматически создавать спаны для функций и связывать логи с контекстом трассировки. Функционал является опциональным и требует установки chutils[otel].

get_current_trace_context()

Возвращает текущие trace_id и span_id, если трассировка активна.

Returns:

Type	Description
Optional[dict[str, str]]	Словарь с trace_id и span_id или None.

get_tracer(name='chutils')

Возвращает экземпляр трейсера, если OpenTelemetry доступен.

Parameters:

Name	Type	Description	Default
name	str	Имя трейсера.	'chutils'

Returns:

Type	Description
Any	Экземпляр Tracer или None, если OTel не установлен.

setup_tracing(service_name, exporter_type='console', otlp_endpoint=None, otlp_protocol='grpc')

Настраивает OpenTelemetry SDK для сбора трасс.

Parameters:

Name	Type	Description	Default
service_name	str	Имя сервиса для отображения в трассах.	required
exporter_type	str	Тип экспортера: 'console' или 'otlp'.	'console'
otlp_endpoint	Optional[str]	URL эндпоинта для OTLP (например, http://localhost:4317).	None
otlp_protocol	str	Протокол для OTLP: 'grpc' или 'http/protobuf'.	'grpc'

Returns:

Type	Description
bool	True, если настройка выполнена успешно, False если OTel недоступен.

trace(name=None, attributes=None, capture_kwargs=False)

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

Если OpenTelemetry не установлен, декоратор просто возвращает оригинальную функцию без накладных расходов.

Parameters:

Name	Type	Description	Default
name	Optional[Any]	Имя спана. По умолчанию используется имя функции. Может использоваться как позиционный аргумент при @trace("имя").	None
attributes	Optional[dict[str, Any]]	Дополнительные атрибуты для спана.	None
capture_kwargs	bool	Если True, аргументы функции будут добавлены как атрибуты спана с префиксом 'arg.'.	False

Example

@trace()
def my_func(x):
    return x + 1

@trace("custom_name", capture_kwargs=True)
async def my_async_func(y):
    return y * 2

options: members:

• trace
• setup_tracing
• IS_OTEL_AVAILABLE

Модуль features (Фича-флаги)

chutils.features

Модуль для управления фича-флагами (Feature Flags).

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

get_features()

Загружает и кэширует фича-флаги.

Приоритет источников: 1. Файл features.yml (или features.yaml) в корне проекта. 2. Секция feature_flags или FeatureFlags в основном config.yml.

Returns:

Type	Description
Dict[str, Any]	Словарь с конфигурацией фича-флагов.

is_feature_enabled(feature_name, context=None)

Проверяет, включена ли указанная фича.

Parameters:

Name	Type	Description	Default
feature_name	str	Уникальное имя фичи.	required
context	Optional[Dict[str, Any]]	Опциональный контекст для вычисления (например, {'user_id': 123}).	None

Returns:

Type	Description
bool	True, если фича включена. False во всех остальных случаях (включая отсутствие фичи).

require_feature(feature_name, fallback=None)

Декоратор для ограничения доступа к функции на основе фича-флага.

Если фича включена, вызывается оригинальная функция. Если выключена: - И задан fallback, вызывается он. - И fallback не задан, возвращается None.

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

Parameters:

Name	Type	Description	Default
feature_name	str	Имя фичи.	required
fallback	Optional[Callable]	Опциональная функция для вызова при выключенной фиче.	None

Returns:

Type	Description
	Декоратор.

options: members:

• is_feature_enabled
• require_feature

Модуль cache (Умное кэширование)

chutils.cache

BaseCacheBackend

Bases: ABC

Базовый абстрактный класс для всех бэкендов кэширования.

Определяет единый интерфейс для синхронных и асинхронных операций.

aclear() async

Асинхронная очистка кэша.

adelete(key) async

Асинхронное удаление значения.

aexists(key) async

Асинхронная проверка наличия ключа.

aget(key) async

Асинхронное получение значения.

aset(key, value, ttl=None) async

Асинхронное сохранение значения.

clear() abstractmethod

Очистить весь кэш.

delete(key) abstractmethod

Удалить ключ из кэша.

Parameters:

Name	Type	Description	Default
key	str	Ключ кэша.	required

exists(key) abstractmethod

Проверить наличие ключа в кэше.

Parameters:

Name	Type	Description	Default
key	str	Ключ кэша.	required

Returns:

Name	Type	Description
bool	bool	True, если ключ существует и не просрочен.

get(key) abstractmethod

Получить значение из кэша.

Parameters:

Name	Type	Description	Default
key	str	Ключ кэша.	required

Returns:

Name	Type	Description
Any	Any	Значение или None, если ключ не найден или просрочен.

set(key, value, ttl=None) abstractmethod

Сохранить значение в кэше.

Parameters:

Name	Type	Description	Default
key	str	Ключ кэша.	required
value	Any	Значение для сохранения.	required
ttl	Optional[int]	Время жизни в секундах. Если None, используется вечное хранение.	None

InMemoryCacheBackend

Bases: BaseCacheBackend

Реализация кэша в оперативной памяти на базе словаря.

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

clear()

Полная очистка.

delete(key)

Удалить ключ.

exists(key)

Проверить наличие ключа (удаляет, если просрочен).

get(key)

Получить значение. Если просрочено - удаляет его.

set(key, value, ttl=None)

Сохранить значение с TTL.

cache_with_ttl(ttl=60, key_prefix='', sliding=True, backend=None)

Декоратор для кэширования результатов выполнения функций с поддержкой TTL.

Parameters:

Name	Type	Description	Default
ttl	int	Время жизни закэшированного значения в секундах. По умолчанию 60.	60
key_prefix	str	Префикс для ключа кэша.	''
sliding	bool	Если True, TTL продлевается при каждом успешном чтении из кэша.	True
backend	Optional[InMemoryCacheBackend]	Инстанс бэкенда для хранения (по умолчанию InMemoryCacheBackend).	None

Returns:

Name	Type	Description
Callable	Callable	Обернутая функция.

options: members:

• cache_with_ttl
• BaseCacheBackend
• InMemoryCacheBackend

Модуль secret_manager

chutils.secret_manager

DotEnvProvider

Bases: SecretProvider

Провайдер для работы с .env файлами. Обеспечивает загрузку переменных окружения из файла при первом обращении.

__init__(dotenv_path=None)

Инициализирует провайдер.

Parameters:

Name	Type	Description	Default
dotenv_path	Optional[str]	Явный путь к .env файлу. Если не указан, ищется в корне проекта.	None

delete(key, service_name)

DotEnvProvider не поддерживает удаление.

get(key, service_name)

Получает значение из загруженных .env данных.

set(key, value, service_name)

DotEnvProvider не поддерживает сохранение (доступен только для чтения).

EnvProvider

Bases: SecretProvider

Провайдер для работы с переменными окружения ОС (os.environ).

delete(key, service_name)

EnvProvider не поддерживает удаление.

get(key, service_name)

Получает значение из переменных окружения ОС.

set(key, value, service_name)

EnvProvider не поддерживает сохранение.

KeyringProvider

Bases: SecretProvider

Провайдер для работы с системным хранилищем (keyring). Использует возможности ОС (Windows Credential Locker, macOS Keychain, KWallet/Secret Service).

__init__(disable_keyring=False)

Инициализирует провайдер.

Parameters:

Name	Type	Description	Default
disable_keyring	bool	Если True, все операции с keyring будут отключены.	False

delete(key, service_name)

Удаляет пароль из системного хранилища.

get(key, service_name)

Получает пароль из системного хранилища.

set(key, value, service_name)

Сохраняет пароль в системное хранилище.

SecretManager

Универсальный менеджер для управления секретами через цепочку провайдеров.

Позволяет получать, сохранять и удалять секреты, используя различные стратегии (Keyring, .env, переменные окружения и т.д.).

__init__(service_name=None, prefix=None, auto_mask_logs=True, providers=None)

Инициализирует менеджер секретов.

Parameters:

Name	Type	Description	Default
service_name	Optional[str]	Уникальное имя сервиса. Если не указано, определяется автоматически.	None
prefix	Optional[str]	Префикс для имени сервиса (по умолчанию "Chutils_").	None
auto_mask_logs	bool	Если True, полученные секреты будут маскироваться в логах.	True
providers	Optional[List[SecretProvider]]	Список провайдеров. Если None, создается стандартная цепочка.	None

Raises:

Type	Description
SecretError	Если не удалось автоматически определить service_name.

add_provider(provider, index=None)

Добавляет новый провайдер в цепочку.

Parameters:

Name	Type	Description	Default
provider	SecretProvider	Экземпляр провайдера.	required
index	Optional[int]	Позиция в списке. Если None, добавляется в конец.	None

delete_secret(key)

Удаляет секрет во всех провайдерах, поддерживающих удаление.

get_secret(key)

Получает секрет, опрашивая провайдеры по порядку.

save_secret(key, value)

Сохраняет секрет в первом провайдере, поддерживающем запись.

update_secret(key, value)

Обновляет секрет (псевдоним для save_secret).

SecretProvider

Bases: ABC

Абстрактный базовый класс для провайдеров секретов. Определяет интерфейс стратегии для различных механизмов хранения.

delete(key, service_name) abstractmethod

Удалить секрет.

Parameters:

Name	Type	Description	Default
key	str	Имя секрета.	required
service_name	str	Имя сервиса.	required

Returns:

Type	Description
bool	True, если удаление прошло успешно, иначе False.

get(key, service_name) abstractmethod

Получить значение секрета по ключу.

Parameters:

Name	Type	Description	Default
key	str	Имя секрета.	required
service_name	str	Имя сервиса (используется для изоляции в хранилищах типа keyring).	required

Returns:

Type	Description
Optional[str]	Значение секрета или None, если секрет не найден.

set(key, value, service_name) abstractmethod

Сохранить значение секрета.

Parameters:

Name	Type	Description	Default
key	str	Имя секрета.	required
value	str	Значение секрета.	required
service_name	str	Имя сервиса.	required

Returns:

Type	Description
bool	True, если сохранение прошло успешно, иначе False.

Модуль config.diagnostics (Отладка конфигурации)

chutils.config.diagnostics

Логика формирования диагностических отчетов по конфигурации.

format_trace(trace_data, format_type='tree', show_secrets=False)

Форматирует данные трассировки в выбранный формат.

mask_value(key, value, show_secrets=False)

Маскирует значение, если ключ похож на секрет.

handler: python

Модуль fs

chutils.fs

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

atomic_write(file_path, data, mode='w', encoding='utf-8', **kwargs)

Атомарная запись данных в файл.

Данные сначала записываются во временный файл в той же директории, после чего выполняется атомарная замена целевого файла (os.replace). Это гарантирует, что файл не будет поврежден при сбое во время записи.

Поддерживает автоматическую сериализацию для JSON и YAML на основе расширения файла.

Parameters:

Name	Type	Description	Default
file_path	Union[str, Path]	Путь к целевому файлу.	required
data	Any	Данные для записи. Может быть строкой, байтами, словарем или списком.	required
mode	str	Режим открытия файла ('w' или 'wb').	'w'
encoding	str	Кодировка (только для текстового режима).	'utf-8'
**kwargs	Any	Дополнительные аргументы для json.dump или yaml.dump.	{}

Raises:

Type	Description
OptionalDependencyError	Если выполняется запись в YAML, но пакет pyyaml не установлен.
OSError	При ошибках ввода-вывода.

ensure_dir(path)

Гарантирует существование директории. Создает все родительские директории, если они не существуют.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	Путь к директории (строка или pathlib.Path).	required

Returns:

Type	Description
Path	Объект pathlib.Path.

get_temp_file(suffix='')

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

Parameters:

Name	Type	Description	Default
suffix	str	Суффикс (расширение) временного файла.	''

Yields:

Type	Description
ContextManager[Path]	Объект pathlib.Path к временному файлу.

options: members:

• ensure_dir
• atomic_write
• get_temp_file

Декораторы

chutils.decorators

Модуль с полезными декораторами для автоматизации задач.

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

log_function_details(func)

Декоратор для логирования деталей вызова функции.

Записывает аргументы, время выполнения и возвращаемое значение на уровне DEVDEBUG.

Parameters:

Name	Type	Description	Default
func	F	Декорируемая функция.	required

Returns:

Type	Description
F	Обертка функции с логированием.

Example

@log_function_details
def add(a, b):
    return a + b

add(2, 3)

retry(retries=3, delay=1.0, backoff=2.0, jitter=False, exceptions=(Exception,))

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

Parameters:

Name	Type	Description	Default
retries	int	Количество попыток повтора (не считая первый запуск).	3
delay	float	Базовая задержка между попытками в секундах.	1.0
backoff	float	Множитель задержки для каждой следующей попытки.	2.0
jitter	bool	Добавлять ли случайный шум к задержке.	False
exceptions	Tuple[Type[Exception], ...]	Кортеж исключений, при которых требуется повтор.	(Exception,)

Returns:

Type	Description
Callable	Декоратор функции.

timeout(seconds, fallback=_NO_FALLBACK)

Декоратор для ограничения времени выполнения функции.

Поддерживает как синхронные, так и асинхронные функции. Для асинхронных функций использует asyncio.wait_for. Для синхронных функций запускает их в отдельном потоке и ожидает завершения.

Parameters:

Name	Type	Description	Default
seconds	float	Максимальное время выполнения в секундах.	required
fallback	Any	Значение, которое будет возвращено при таймауте. Если не указано, выбрасывается ChutilsTimeoutError.	_NO_FALLBACK

Returns:

Type	Description
Callable	Декоратор функции.

Raises:

Type	Description
ChutilsTimeoutError	Если время выполнения превышено и fallback не указан.

options: members:

• retry
• log_function_details
• timeout

Исключения

chutils.exceptions

CacheError

Bases: ChutilsException

Общая ошибка кэширования.

ChutilsException

Bases: Exception

Базовый класс для всех исключений библиотеки chutils.

Поддерживает структурированный контекст ошибки через именованные аргументы.

ChutilsTimeoutError

Bases: ChutilsException

Ошибка: превышено время ожидания выполнения операции.

ConfigError

Bases: ChutilsException

Общая ошибка конфигурации.

ConfigLoadError

Bases: ConfigError

Ошибка при загрузке файла конфигурации (отсутствие файла, права доступа).

ConfigParseError

Bases: ConfigError

Ошибка при парсинге содержимого конфигурации (невалидный YAML/JSON/INI).

LoggerConfigurationError

Bases: ChutilsException

Ошибка конфигурации логгера.

OptionalDependencyError

Bases: ChutilsException

Ошибка: отсутствует опциональная зависимость (например, watchdog).

SecretError

Bases: ChutilsException

Общая ошибка менеджера секретов.

SecretNotFoundError

Bases: SecretError

Ошибка: секрет не найден.

SecretProviderError

Bases: SecretError

Ошибка конкретного провайдера секретов (например, сбой keyring).

WatcherInitializationError

Bases: ChutilsException

Ошибка инициализации наблюдателя (watcher) за файлами.

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

Подробную информацию о pytest-фикстурах для тестирования приложений с chutils см. в разделе Тестирование с chutils.

chutils.testing

capture_chutils_logs()

Фикстура для перехвата логов.

• Перехватывает все логи, проходящие через любой логгер (включая те, где propagate=False).
• Позволяет проверять сообщения и поля контекста (например, добавленные через bind_context).

Example

def test_logging(capture_chutils_logs): from chutils.logger import setup_logger logger = setup_logger("test") logger.info("Hello world") assert capture_chutils_logs.has_message("Hello")

mock_chutils_config(monkeypatch)

Фикстура для мокирования конфигурации chutils.

• Отключает переопределение через переменные окружения (CH_DISABLE_ENV_OVERRIDE=true).
• Сбрасывает состояние глобального ConfigManager до и после теста.
• Возвращает объект с методом .set(section, key, value).

mock_chutils_secrets(monkeypatch)

Фикстура для мокирования секретов chutils.

• Заменяет все провайдеры в SecretManager на один MockSecretProvider.
• Отключает предупреждение о миграции keyring.