Справочник 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) должны находиться в той же директории, что и основной файл. Это позволяет удобно управлять чувствительными или специфичными для разработчика настройками, не коммитя их в репозиторий.

BaseConfigProvider

Bases: ABC

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

Реализуйте этот класс, чтобы интегрировать любой внешний источник настроек (БД, Redis, Consul, удалённый API и т.д.) в систему конфигурации chutils.

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

Приоритет между несколькими провайдерами определяется параметром priority при регистрации: меньшее число → больший приоритет (аналогично DNS TTL).

Example

Минимальная реализация синхронного провайдера::

class EnvVaultProvider(BaseConfigProvider):
    def get_value(self, section: str, key: str) -> Any | None:
        return vault_client.get(f"{section}/{key}")

    async def aget_value(self, section: str, key: str) -> Any | None:
        return await async_vault_client.get(f"{section}/{key}")

aget_value(section, key) abstractmethod async

Асинхронно получает значение из провайдера.

Parameters:

Name	Type	Description	Default
section	str	Имя секции конфигурации.	required
key	str	Имя ключа внутри секции.	required

Returns:

Type	Description
Any | None	Значение из источника или None, если ключ не найден.

get_value(section, key) abstractmethod

Синхронно получает значение из провайдера.

Parameters:

Name	Type	Description	Default
section	str	Имя секции конфигурации.	required
key	str	Имя ключа внутри секции.	required

Returns:

Type	Description
Any | None	Значение из источника или None, если ключ не найден.

DictConfigProvider

Bases: BaseConfigProvider

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

Предназначен для использования в тестах (моки настроек) и для быстрого прототипирования. Полностью потокобезопасен при чтении, так как словарь не изменяется после инициализации.

Example

Использование в тестах с pytest::

import pytest
from chutils.config import register_provider, reset_providers
from chutils.config.custom_providers import DictConfigProvider

@pytest.fixture(autouse=True)
def mock_config():
    provider = DictConfigProvider({
        "database": {"host": "test-host", "port": "5432"},
        "app": {"debug": "true"},
    })
    register_provider(provider, priority=0)
    yield
    reset_providers()

Attributes:

Name	Type	Description
_data	dict[str, dict[str, Any]]	Вложенный словарь вида {section: {key: value}}.

__init__(data)

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

Parameters:

Name	Type	Description	Default
data	dict[str, dict[str, Any]]	Вложенный словарь вида {section: {key: value}}. Ключи секций и ключей сравниваются без учёта регистра.	required

aget_value(section, key) async

Асинхронно возвращает значение из словаря.

Реализация не блокирует event loop, поскольку работает только с данными в памяти.

Parameters:

Name	Type	Description	Default
section	str	Имя секции (без учёта регистра).	required
key	str	Имя ключа (без учёта регистра).	required

Returns:

Type	Description
Any | None	Значение или None, если секция или ключ не найдены.

get_value(section, key)

Синхронно возвращает значение из словаря.

Parameters:

Name	Type	Description	Default
section	str	Имя секции (без учёта регистра).	required
key	str	Имя ключа (без учёта регистра).	required

Returns:

Type	Description
Any | None	Значение или None, если секция или ключ не найдены.

__getattr__(name)

Обеспечивает ленивую загрузку экспортируемых функций генератора и схемы.

aget_config(model=None) async

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

Parameters:

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

Returns:

Type	Description
JSONDict | T	Словарь конфигурации или экземпляр Pydantic модели.

aget_config_value(section, key, fallback=None, config=None, required=False) async

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

Сначала опрашивает зарегистрированные кастомные провайдеры через их асинхронный интерфейс aget_value. Если ни один из них не вернул значение, выполняет синхронный поиск в уже загруженном кэше конфигурации (через run_in_executor, чтобы не блокировать event loop).

Parameters:

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

Returns:

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

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	str | None	Опциональный путь к файлу для сохранения. Если указан, имеет приоритет над 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	type[BaseModel] | str	Класс модели или строковый путь к нему ('module:Class').	required
output_path	str | Path | None	Путь к файлу для сохранения схемы.	None
indent	int	Отступ в результирующем JSON.	4

Returns:

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

generate_env_template(model_class, prefix='CH')

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

Parameters:

Name	Type	Description	Default
model_class	type[BaseModel]	Класс Pydantic модели.	required
prefix	str	Префикс для имен переменных окружения.	'CH'

Returns:

Type	Description
str	Строковое представление сгенерированного .env-шаблона.

generate_json_schema(model_class)

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

Parameters:

Name	Type	Description	Default
model_class	type[BaseModel]	Класс Pydantic модели.	required

Returns:

Type	Description
str	JSON-строка схемы.

generate_yaml_template(model_class, indent=0)

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

Parameters:

Name	Type	Description	Default
model_class	type[BaseModel]	Класс Pydantic модели.	required
indent	int	Начальный отступ в пробелах (уровень вложенности).	0

Returns:

Type	Description
str	Строковое представление сгенерированного YAML-шаблона.

get_all_config_paths(cfg_file=None)

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

Parameters:

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

Returns:

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

get_base_dir()

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

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

Returns:

Type	Description
str | None	Путь к корню проекта или 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	type[T] | None	Опциональный класс Pydantic модели для валидации.	None
remote_url	str | None	URL для загрузки удаленной конфигурации.	None
remote_auth	tuple[str, str] | None	Кортеж (login, password) для Basic Auth.	None
polling_interval	int | None	Интервал опроса удаленного источника в секундах. Если не указан, опрос не запускается.	None

Returns:

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

Raises:

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

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

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

Распознает '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	JSONDict | None	Опциональный, предварительно загруженный словарь конфигурации.	None
required	bool	Если True, выбросит ConfigKeyNotFoundError при отсутствии ключа.	False

Returns:

Type	Description
bool	True или False.

get_config_file_path()

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

Returns:

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

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

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

Parameters:

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

Returns:

Type	Description
float	Float или fallback.

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

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

Parameters:

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

Returns:

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

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

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

Parameters:

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

Returns:

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

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

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

get_config_paths(cfg_file=None)

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

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

Parameters:

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

Returns:

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

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

get_config_section(
    section_name: str,
    fallback: JSONDict | None = None,
    config: JSONDict | None = None,
    model: None = None,
    required: bool = False,
) -> JSONDict

get_config_section(
    section_name: str,
    fallback: JSONDict | None = None,
    config: JSONDict | None = None,
    model: type[T] = ...,
    required: bool = False,
) -> T

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

Parameters:

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

Returns:

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

Raises:

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

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

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

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

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

1. Зарегистрированные кастомные провайдеры (по их приоритету).
2. Переменные окружения CH_[SECTION]_[KEY].
3. Локальный файл конфигурации (config.local.yml).
4. Файл окружения (config.{CH_ENV}.yml).
5. Основной файл конфигурации (config.yml).

Parameters:

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

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
ConfigParseError	Если формат пути некорректен или класс не найден.
OptionalDependencyError	Если Pydantic не установлен.

is_config_loaded()

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

Returns:

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

load_ai_lint_config(cli_args=None)

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

Приоритет источников (от наивысшего к низшему): 1. CLI флаги (cli_args) 2. Переменные окружения (CH_DEV_AILINT_...) 3. Локальные yml файлы (секция Dev.AI-Lint) 4. pyproject.toml (секция [tool.chutils.ai-lint]) 5. Значения по умолчанию

Parameters:

Name	Type	Description	Default
cli_args	JSONDict | None	Аргументы командной строки.	None

Returns:

Type	Description
JSONDict	Объединенный словарь конфигурации ai-lint.

on_config_change(callback)

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

Parameters:

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

parse_chutils_ignore(base_dir)

Парсит файл .chutilsignore и возвращает список шаблонов для игнорирования.

Parameters:

Name	Type	Description	Default
base_dir	str	Корневая директория, содержащая .chutilsignore.	required

Returns:

Type	Description
list[str]	Список шаблонов для игнорирования.

register_provider(provider, priority=100)

Регистрирует кастомный провайдер конфигурации.

Провайдеры опрашиваются перед чтением локальных файлов конфигурации. Если провайдер возвращает значение (не None), оно используется как итоговое.

Приоритет: меньшее число → выше приоритет (опрашивается первым).

Parameters:

Name	Type	Description	Default
provider	BaseConfigProvider	Экземпляр класса, реализующего :class:BaseConfigProvider.	required
priority	int	Числовой приоритет провайдера. По умолчанию: 100.	100

Example

::

from chutils.config import register_provider
from chutils.config.custom_providers import DictConfigProvider

provider = DictConfigProvider({"db": {"host": "prod-db"}})
register_provider(provider, priority=10)

reset_providers()

Очищает реестр всех зарегистрированных кастомных провайдеров.

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

Example

::

from chutils.config import reset_providers

def teardown():
    reset_providers()

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	str | None	Опциональный путь к файлу для сохранения. Если указан, имеет приоритет над 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()

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

validate_required_keys(section, keys, config=None)

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

Parameters:

Name	Type	Description	Default
section	str	Имя секции для валидации.	required
keys	list[str]	Список ключей, которые должны присутствовать и быть не пустыми.	required
config	JSONDict | None	Опциональный, предварительно загруженный словарь конфигурации.	None

Raises:

Type	Description
ConfigValidationGroupError	Если один или несколько ключей отсутствуют.

options: members:

• get_config
• aget_config
• get_config_value
• aget_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
• register_provider
• reset_providers
• BaseConfigProvider
• DictConfigProvider

Модуль config.custom_providers (Custom Config Providers API)

chutils.config.custom_providers

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

Позволяет регистрировать внешние источники настроек (БД, Key-Value хранилища, удалённые API) и интегрировать их в единый интерфейс chutils.config.

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

from chutils.config import register_provider
from chutils.config.custom_providers import DictConfigProvider

provider = DictConfigProvider({"database": {"host": "localhost"}})
register_provider(provider, priority=10)

BaseConfigProvider

Bases: ABC

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

Реализуйте этот класс, чтобы интегрировать любой внешний источник настроек (БД, Redis, Consul, удалённый API и т.д.) в систему конфигурации chutils.

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

Приоритет между несколькими провайдерами определяется параметром priority при регистрации: меньшее число → больший приоритет (аналогично DNS TTL).

Example

Минимальная реализация синхронного провайдера::

class EnvVaultProvider(BaseConfigProvider):
    def get_value(self, section: str, key: str) -> Any | None:
        return vault_client.get(f"{section}/{key}")

    async def aget_value(self, section: str, key: str) -> Any | None:
        return await async_vault_client.get(f"{section}/{key}")

aget_value(section, key) abstractmethod async

Асинхронно получает значение из провайдера.

Parameters:

Name	Type	Description	Default
section	str	Имя секции конфигурации.	required
key	str	Имя ключа внутри секции.	required

Returns:

Type	Description
Any | None	Значение из источника или None, если ключ не найден.

get_value(section, key) abstractmethod

Синхронно получает значение из провайдера.

Parameters:

Name	Type	Description	Default
section	str	Имя секции конфигурации.	required
key	str	Имя ключа внутри секции.	required

Returns:

Type	Description
Any | None	Значение из источника или None, если ключ не найден.

DictConfigProvider

Bases: BaseConfigProvider

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

Предназначен для использования в тестах (моки настроек) и для быстрого прототипирования. Полностью потокобезопасен при чтении, так как словарь не изменяется после инициализации.

Example

Использование в тестах с pytest::

import pytest
from chutils.config import register_provider, reset_providers
from chutils.config.custom_providers import DictConfigProvider

@pytest.fixture(autouse=True)
def mock_config():
    provider = DictConfigProvider({
        "database": {"host": "test-host", "port": "5432"},
        "app": {"debug": "true"},
    })
    register_provider(provider, priority=0)
    yield
    reset_providers()

Attributes:

Name	Type	Description
_data	dict[str, dict[str, Any]]	Вложенный словарь вида {section: {key: value}}.

__init__(data)

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

Parameters:

Name	Type	Description	Default
data	dict[str, dict[str, Any]]	Вложенный словарь вида {section: {key: value}}. Ключи секций и ключей сравниваются без учёта регистра.	required

aget_value(section, key) async

Асинхронно возвращает значение из словаря.

Реализация не блокирует event loop, поскольку работает только с данными в памяти.

Parameters:

Name	Type	Description	Default
section	str	Имя секции (без учёта регистра).	required
key	str	Имя ключа (без учёта регистра).	required

Returns:

Type	Description
Any | None	Значение или None, если секция или ключ не найдены.

get_value(section, key)

Синхронно возвращает значение из словаря.

Parameters:

Name	Type	Description	Default
section	str	Имя секции (без учёта регистра).	required
key	str	Имя ключа (без учёта регистра).	required

Returns:

Type	Description
Any | None	Значение или None, если секция или ключ не найдены.

get_registry()

Возвращает глобальный реестр кастомных провайдеров.

Returns:

Type	Description
_CustomProviderRegistry	Глобальный экземпляр :class:_CustomProviderRegistry.

options: members:

• BaseConfigProvider
• DictConfigProvider

Модуль logger

chutils.logger

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

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

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-логирования).

filter(record)

Обогащает запись лога контекстными данными.

Parameters:

Name	Type	Description	Default
record	LogRecord	Запись лога, которую необходимо отфильтровать/обогатить.	required

Returns:

Type	Description
bool	Всегда возвращает True для продолжения обработки записи.

bind_context(**kwargs)

Привязывает значения к текущему контексту.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Ключи и значения для привязки к контексту.	{}

Returns:

Type	Description
Token[dict[str, Any]]	Токен для последующей очистки контекста через unbind_context.

clear_context()

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

get_context()

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

Returns:

Type	Description
dict[str, Any]	Словарь с текущими контекстными переменными.

unbind_context(token)

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

Parameters:

Name	Type	Description	Default
token	Token[dict[str, Any]]	Токен, возвращенный соответствующим вызовом bind_context.	required

options: members:

• bind_context
• unbind_context
• clear_context
• ContextFilter

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

chutils.lifecycle

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

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

LifecycleManager

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

__init__()

Инициализирует LifecycleManager.

get_cleanup_callbacks()

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

Returns:

Type	Description
list[CleanupCallback]	Список функций очистки в порядке 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)

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

Parameters:

Name	Type	Description	Default
signals	list[int] | None	Опциональный список сигналов для отслеживания.	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) для генерации справки.

Parameters:

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

Returns:

Type	Description
F	Обертка функции, поддерживающая CLI-интерфейс.

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	dict[str, Any] | None	Дополнительные локали или переопределения.	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	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

IS_OTEL_AVAILABLE = OTEL_AVAILABLE module-attribute

Флаг доступности OpenTelemetry, используемый для обратной совместимости.

get_current_trace_context()

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

Returns:

Type	Description
dict[str, str] | None	Словарь с 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	str | None	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	str | Callable[P, R] | None	Имя спана. По умолчанию используется имя функции. Может использоваться как позиционный аргумент при @trace("имя").	None
attributes	dict[str, Any] | None	Дополнительные атрибуты для спана.	None
capture_kwargs	bool	Если True, аргументы функции будут добавлены как атрибуты спана с префиксом 'arg.'.	False

Returns:

Type	Description
Any	Декоратор для создания спана или саму декорируемую функцию.

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	dict[str, Any] | None	Опциональный контекст для вычисления (например, {'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	Callable[..., Any] | None	Опциональная функция для вызова при выключенной фиче.	None

Returns:

Type	Description
Callable[..., Any]	Декоратор.

options: members:

• is_feature_enabled
• require_feature

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

chutils.cache

BaseCacheBackend

Bases: ABC, Generic[T]

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

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

aclear() async

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

adelete(key) async

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

Parameters:

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

aexists(key) async

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

Parameters:

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

Returns:

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

aget(key) async

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

Parameters:

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

Returns:

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

ainvalidate_tag(tag) async

Асинхронное удаление всех ключей, связанных с указанным тегом.

Parameters:

Name	Type	Description	Default
tag	str	Тег для инвалидации.	required

aset(key, value, ttl=None, tags=None) async

Асинхронное сохранение значения в кэше.

Parameters:

Name	Type	Description	Default
key	str	Ключ кэша.	required
value	T	Значение для сохранения.	required
ttl	int | None	Время жизни в секундах.	None
tags	list[str] | None	Список тегов для связывания с ключом.	None

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:

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

invalidate_tag(tag) abstractmethod

Удалить все ключи, связанные с указанным тегом.

Parameters:

Name	Type	Description	Default
tag	str	Тег для инвалидации.	required

set(key, value, ttl=None, tags=None) abstractmethod

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

Parameters:

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

InMemoryCacheBackend

Bases: BaseCacheBackend[T]

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

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

__init__()

Инициализирует бэкенд кэширования в памяти.

aclear() async

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

adelete(key) async

Асинхронно удаляет запись из кэша по ключу.

Parameters:

Name	Type	Description	Default
key	str	Ключ для удаления.	required

aexists(key) async

Асинхронно проверяет существование ключа в кэше.

Parameters:

Name	Type	Description	Default
key	str	Ключ для проверки.	required

Returns:

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

aget(key) async

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

Parameters:

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

Returns:

Type	Description
T | None	Значение из кэша или None, если оно отсутствует или просрочено.

ainvalidate_tag(tag) async

Асинхронно удаляет все ключи, связанные с указанным тегом.

Parameters:

Name	Type	Description	Default
tag	str	Тег для инвалидации.	required

aset(key, value, ttl=None, tags=None) async

Асинхронно сохраняет значение с TTL и тегами.

Parameters:

Name	Type	Description	Default
key	str	Ключ кэша.	required
value	T	Сохраняемое значение.	required
ttl	int | None	Время жизни записи в секундах.	None
tags	list[str] | None	Список тегов для связывания с ключом.	None

clear()

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

delete(key)

Удаляет запись из кэша по ключу.

Parameters:

Name	Type	Description	Default
key	str	Ключ для удаления.	required

exists(key)

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

Parameters:

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

Returns:

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

get(key)

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

Parameters:

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

Returns:

Type	Description
T | None	Значение из кэша или None, если оно отсутствует или просрочено.

invalidate_tag(tag)

Удаляет все ключи, связанные с указанным тегом.

Parameters:

Name	Type	Description	Default
tag	str	Тег для инвалидации.	required

set(key, value, ttl=None, tags=None)

Сохраняет значение с заданным TTL и тегами.

Parameters:

Name	Type	Description	Default
key	str	Ключ кэша.	required
value	T	Сохраняемое значение.	required
ttl	int | None	Время жизни записи в секундах.	None
tags	list[str] | None	Список тегов для связывания с ключом.	None

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

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

Декорированная функция получает дополнительные методы управления кэшем: * invalidate(*args, **kwargs) / ainvalidate(*args, **kwargs) — удаляет запись из кэша для указанных аргументов. * invalidate_all() / ainvalidate_all() — полностью очищает все записи этой функции. * invalidate_tag(tag) / ainvalidate_tag(tag) — сбрасывает все записи кэша, помеченные данным тегом.

Parameters:

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

Returns:

Name	Type	Description
Callable	Callable[..., Any]	Обернутая функция со встроенными методами инвалидации.

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	str | None	Явный путь к .env файлу. Если не указан, ищется в корне проекта.	None

delete(key, service_name)

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

Parameters:

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

Returns:

Type	Description
bool	Всегда False, так как удаление не поддерживается.

get(key, service_name)

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

Parameters:

Name	Type	Description	Default
key	str	Имя запрашиваемого секрета.	required
service_name	str	Имя сервиса/приложения.	required

Returns:

Type	Description
str | None	Значение секрета или None, если он не найден.

set(key, value, service_name)

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

Parameters:

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

Returns:

Type	Description
bool	Всегда False, так как запись не поддерживается.

EnvProvider

Bases: SecretProvider

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

delete(key, service_name)

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

Parameters:

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

Returns:

Type	Description
bool	Всегда False, так как удаление не поддерживается.

get(key, service_name)

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

Parameters:

Name	Type	Description	Default
key	str	Имя запрашиваемого секрета.	required
service_name	str	Имя сервиса/приложения.	required

Returns:

Type	Description
str | None	Значение секрета или None, если он не найден.

set(key, value, service_name)

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

Parameters:

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

Returns:

Type	Description
bool	Всегда False, так как запись не поддерживается.

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)

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

Parameters:

Name	Type	Description	Default
key	str	Имя удаляемого секрета.	required
service_name	str	Имя сервиса/приложения.	required

Returns:

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

get(key, service_name)

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

Parameters:

Name	Type	Description	Default
key	str	Имя запрашиваемого секрета.	required
service_name	str	Имя сервиса/приложения.	required

Returns:

Type	Description
str | None	Значение секрета или None, если он не найден.

set(key, value, service_name)

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

Parameters:

Name	Type	Description	Default
key	str	Имя секрета.	required
value	str	Сохраняемое значение секрета.	required
service_name	str	Имя сервиса/приложения.	required

Returns:

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

SecretManager

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

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

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

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

Parameters:

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

Raises:

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

add_provider(provider, index=None)

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

Parameters:

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

adelete_secret(key) async

Асинхронно удаляет секрет.

Parameters:

Name	Type	Description	Default
key	str	Имя удаляемого секрета.	required

Returns:

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

aget_secret(key, fallback=None, required=False) async

Асинхронно получает секрет.

Parameters:

Name	Type	Description	Default
key	str	Имя секрета.	required
fallback	str | None	Значение по умолчанию.	None
required	bool	Флаг обязательности.	False

Returns:

Type	Description
str | None	Значение секрета или fallback.

asave_secret(key, value) async

Асинхронно сохраняет секрет.

Parameters:

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

Returns:

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

delete_secret(key)

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

Parameters:

Name	Type	Description	Default
key	str	Имя удаляемого секрета.	required

Returns:

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

get_secret(key, fallback=None, required=False)

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

Parameters:

Name	Type	Description	Default
key	str	Имя запрашиваемого секрета.	required
fallback	str | None	Значение по умолчанию, если секрет не найден.	None
required	bool	Флаг обязательности. Если True, выбрасывает исключение SecretNotFoundError при отсутствии.	False

Returns:

Type	Description
str | None	Значение секрета или fallback.

save_secret(key, value)

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

Parameters:

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

Returns:

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

update_secret(key, value)

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

Parameters:

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

Returns:

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

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
str | None	Значение секрета или 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)

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

Parameters:

Name	Type	Description	Default
trace_data	dict[str, dict[str, list[dict[str, Any]]]]	Данные трассировки конфигурации.	required
format_type	str	Тип форматирования ('tree', 'table', 'json').	'tree'
show_secrets	bool	Флаг принудительного показа секретов без маскирования.	False

Returns:

Type	Description
str	Отформатированная строка с диагностическим отчетом.

mask_value(key, value, show_secrets=False)

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

Parameters:

Name	Type	Description	Default
key	str	Имя ключа.	required
value	Any	Значение ключа.	required
show_secrets	bool	Флаг принудительного показа секретов без маскирования.	False

Returns:

Type	Description
str	Строковое представление значения (возможно маскированное).

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	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	При ошибках ввода-вывода.

cleanup_paths(*paths, retries=3, delay=0.1, on_locked='warn', orphan_collision='raise')

Пакетное удаление нескольких путей.

Вызывает remove_path для каждого пути. Если on_locked == "warn" или "rename_orphan", ошибки для отдельных путей не прерывают обход остальных. При "raise" выполнение прерывается на первой же ошибке после исчерпания попыток.

Parameters:

Name	Type	Description	Default
paths	str | Path	Пути к удаляемым объектам.	()
retries	int	Количество повторных попыток.	3
delay	float	Задержка между попытками в секундах.	0.1
on_locked	Literal['raise', 'rename_orphan', 'warn']	Поведение при блокировке.	'warn'
orphan_collision	Literal['raise', 'overwrite', 'unique']	Поведение при коллизиях orphan-файлов.	'raise'

Raises:

Type	Description
OSError	Если удаление не удалось и on_locked == "raise".

ensure_dir(path)

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

Parameters:

Name	Type	Description	Default
path	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
Path	Объект pathlib.Path к временному файлу.

remove_path(path, *, retries=3, delay=0.1, on_locked='warn', orphan_collision='raise')

Безопасно удаляет файл или директорию по указанному пути.

При возникновении ошибок доступа осуществляет повторные попытки с задержкой. В случае окончательной блокировки обрабатывает ошибку в соответствии с параметром on_locked.

Parameters:

Name	Type	Description	Default
path	str | Path	Путь к удаляемому объекту.	required
retries	int	Количество повторных попыток.	3
delay	float	Задержка между попытками в секундах.	0.1
on_locked	Literal['raise', 'rename_orphan', 'warn']	Поведение при блокировке ("raise", "rename_orphan", "warn").	'warn'
orphan_collision	Literal['raise', 'overwrite', 'unique']	Поведение при коллизиях orphan-файлов ("raise", "overwrite", "unique").	'raise'

Returns:

Type	Description
bool	True, если объект успешно удален или переименован,
bool	False, если удаление не удалось и on_locked == "warn".

Raises:

Type	Description
OSError	Если удаление не удалось и on_locked == "raise".
FileExistsError	Если orphan-файл существует и orphan_collision == "raise".

resolve_safe_path(path, base_dir=None)

Безопасно разрешает путь относительно базовой директории. Проверяет попытки выхода за пределы базовой директории (Path Traversal).

Parameters:

Name	Type	Description	Default
path	str | Path	Путь для разрешения.	required
base_dir	str | Path | None	Базовая директория. Если не указана, используется корень проекта из конфига.	None

Returns:

Type	Description
Path	Разрешенный абсолютный путь (pathlib.Path).

Raises:

Type	Description
PathTraversalError	Если обнаружена попытка выхода за пределы base_dir.

safe_filename(name, replacement='_', strip_chars=' _.-', max_length=255, transliterate=False)

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

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

Parameters:

Name	Type	Description	Default
name	str	Исходное имя.	required
replacement	str	Символ замены недопустимых символов.	'_'
strip_chars	str	Символы, удаляемые с краев строки.	' _.-'
max_length	int	Максимальная длина результирующей строки.	255
transliterate	bool	Флаг транслитерации кириллицы в латиницу.	False

Returns:

Type	Description
str	Безопасное имя файла.

Raises:

Type	Description
ValueError	Если имя пустое или после очистки получается пустая строка.

zip_folder(folder_path, output_path, compression=zipfile.ZIP_DEFLATED, exclude=None)

Архивирует содержимое папки в ZIP-архив с сохранением структуры.

Parameters:

Name	Type	Description	Default
folder_path	str | Path	Путь к архивируемой папке.	required
output_path	str | Path	Путь к создаваемому ZIP-файлу.	required
compression	int	Метод сжатия (по умолчанию ZIP_DEFLATED).	ZIP_DEFLATED
exclude	list[str] | None	Список glob-шаблонов для исключения файлов/папок.	None

Returns:

Type	Description
Path	Путь к созданному ZIP-архиву.

Raises:

Type	Description
FileNotFoundError	Если папка folder_path не существует.
ValueError	Если folder_path не является директорией.

options: members:

• ensure_dir
• atomic_write
• get_temp_file

Декораторы

chutils.decorators

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

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

CircuitBreakerState

Машина состояний для паттерна Circuit Breaker (Предохранитель).

__init__(failure_threshold, recovery_timeout, exceptions, name)

Инициализирует состояние предохранителя (Circuit Breaker).

Parameters:

Name	Type	Description	Default
failure_threshold	int	Порог неудачных попыток для открытия цепи.	required
recovery_timeout	float	Таймаут восстановления в секундах.	required
exceptions	tuple[type[Exception], ...]	Исключения, расцениваемые как ошибки.	required
name	str	Имя предохранителя (обычно имя декорируемой функции).	required

can_execute()

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

Returns:

Type	Description
bool	True, если выполнение разрешено, иначе False.

record_failure(exc)

Записывает неудачную попытку выполнения.

Parameters:

Name	Type	Description	Default
exc	Exception	Возникшее исключение.	required

record_success()

Записывает успешное выполнение и сбрасывает состояние сбоев в CLOSED.

LeakyBucket

Алгоритм дырявого ведра (Leaky Bucket).

__init__(capacity, period)

Инициализирует дырявое ведро.

Parameters:

Name	Type	Description	Default
capacity	int	Максимальная вместимость ведра (максимальный уровень воды).	required
period	float	Временной интервал в секундах, за который ведро полностью опустошается.	required

acquire(wait=False)

Запрашивает добавление единицы воды в ведро.

Parameters:

Name	Type	Description	Default
wait	bool	Флаг необходимости блокирующего ожидания до тех пор, пока уровень воды не спадет.	False

Returns:

Type	Description
float | None	Время ожидания в секундах, если необходимо подождать, 0.0 если вода добавлена сразу,
float | None	или None, если ведро переполнено и wait=False.

TokenBucket

Алгоритм маркерной корзины (Token Bucket).

__init__(capacity, period)

Инициализирует маркерную корзину.

Parameters:

Name	Type	Description	Default
capacity	int	Максимальная вместимость корзины (количество токенов).	required
period	float	Временной интервал в секундах, за который корзина полностью пополняется.	required

acquire(wait=False)

Запрашивает получение токена из корзины.

Parameters:

Name	Type	Description	Default
wait	bool	Флаг необходимости блокирующего ожидания, если токенов нет.	False

Returns:

Type	Description
float | None	Время ожидания в секундах, если необходимо подождать, 0.0 если токен получен сразу,
float | None	или None, если токенов нет и wait=False.

bulkhead(max_concurrent, max_waiting=0, timeout=None, fallback=_NO_FALLBACK, key=None)

Декоратор для изоляции ресурсов (Bulkhead).

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

Parameters:

Name	Type	Description	Default
max_concurrent	int	Максимальное количество параллельно выполняющихся запросов.	required
max_waiting	int	Максимальное количество запросов в очереди ожидания.	0
timeout	Optional[float]	Таймаут ожидания свободного слота в секундах.	None
fallback	Any	Значение или callable для возврата при отклонении.	_NO_FALLBACK
key	Optional[Callable[..., Any]]	Опциональная функция вычисления динамического ключа группировки.	None

Returns:

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	Декорированная функция.

circuit_breaker(failure_threshold=5, recovery_timeout=60.0, exceptions=(Exception,))

Декоратор Circuit Breaker (Предохранитель) для защиты от каскадных сбоев.

Parameters:

Name	Type	Description	Default
failure_threshold	int	Количество последовательных ошибок для открытия цепи.	5
recovery_timeout	float	Время в секундах, в течение которого цепь остается открытой.	60.0
exceptions	tuple[type[Exception], ...]	Кортеж исключений, которые считаются ошибками.	(Exception,)

Returns:

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	Декоратор, оборачивающий функцию механизмом автоматического отключения при сбоях.

clear_limiters()

Очищает реестр ограничителей (для тестов).

get_limiter(key, max_calls, period, strategy='token_bucket')

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

Parameters:

Name	Type	Description	Default
key	str	Уникальный ключ для идентификации ограничителя.	required
max_calls	int	Максимальное число вызовов за период.	required
period	float	Временной интервал в секундах.	required
strategy	str	Стратегия ограничения частоты ("token_bucket" или "leaky_bucket").	'token_bucket'

Returns:

Type	Description
TokenBucket | LeakyBucket	Экземпляр ограничителя частоты (TokenBucket или LeakyBucket).

log_function_details(func)

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

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

Parameters:

Name	Type	Description	Default
func	Callable[P, R]	Декорируемая функция.	required

Returns:

Type	Description
Callable[P, R]	Обертка функции с логированием.

Example

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

add(2, 3)

rate_limit(max_calls, period, strategy='token_bucket', wait=False, key_func=None)

Декоратор для ограничения частоты вызовов функции (Throttling).

Parameters:

Name	Type	Description	Default
max_calls	int	Максимальное количество вызовов в период.	required
period	float	Период времени в секундах.	required
strategy	str	Стратегия лимитирования ("token_bucket" или "leaky_bucket").	'token_bucket'
wait	bool	Если True, блокирует выполнение до появления токена. Если False, сразу выбрасывает RateLimitExceededError при превышении лимита.	False
key_func	Callable[..., str] | None	Кастомная функция для генерации ключа лимитирования на основе аргументов.	None

Returns:

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	Декоратор функции.

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[[Callable[P, R]], Callable[P, R]]	Декоратор функции.

semaphore(max_concurrent, key=None)

Декоратор для ограничения максимального количества параллельных вызовов функции (Semaphore).

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

Parameters:

Name	Type	Description	Default
max_concurrent	int	Максимальное количество параллельных вызовов.	required
key	Optional[Callable[..., Any]]	Опциональная функция для вычисления динамического ключа группировки на основе аргументов функции.	None

Returns:

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	Декорированная функция.

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[[Callable[P, R]], Callable[P, R]]	Декоратор функции.

Raises:

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

options: members:

• retry
• log_function_details
• timeout
• rate_limit
• circuit_breaker

Модуль events (Шина событий)

chutils.events

Модуль шины событий (In-Memory Event Bus).

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

ErrorStrategy

Bases: str, Enum

Стратегия обработки ошибок при выполнении обработчиков событий.

EventBus

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

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

__init__(error_strategy=ErrorStrategy.IGNORE)

Инициализирует шину событий.

Parameters:

Name	Type	Description	Default
error_strategy	ErrorStrategy	Стратегия обработки ошибок по умолчанию.	IGNORE

publish(event_name, *args, error_strategy=None, **kwargs)

Синхронно публикует событие.

Синхронные обработчики выполняются немедленно в текущем потоке. Асинхронные обработчики запускаются в фоновом режиме в выделенном Event Loop.

Parameters:

Name	Type	Description	Default
event_name	str	Имя события.	required
*args	Any	Позиционные аргументы для обработчиков.	()
error_strategy	ErrorStrategy | None	Стратегия обработки ошибок для этого вызова.	None
**kwargs	Any	Именованные аргументы для обработчиков.	{}

publish_async(event_name, *args, error_strategy=None, **kwargs) async

Асинхронно публикует событие.

Дожидается выполнения всех подписчиков (как синхронных, так и асинхронных). Синхронные обработчики выполняются в пуле потоков через asyncio.to_thread.

Parameters:

Name	Type	Description	Default
event_name	str	Имя события.	required
*args	Any	Позиционные аргументы для обработчиков.	()
error_strategy	ErrorStrategy | None	Стратегия обработки ошибок для этого вызова.	None
**kwargs	Any	Именованные аргументы для обработчиков.	{}

subscribe(event_name)

Декоратор для регистрации обработчика события на данном инстансе шины.

Parameters:

Name	Type	Description	Default
event_name	str	Имя события, на которое подписывается обработчик.	required

Returns:

Type	Description
Callable[[Callable[..., Any]], Callable[..., Any]]	Декоратор, который регистрирует функцию-обработчик и возвращает её.

unsubscribe(event_name, func)

Отменяет подписку обработчика на событие.

Parameters:

Name	Type	Description	Default
event_name	str	Имя события.	required
func	Callable[..., Any]	Функция-обработчик, которую нужно отписать.	required

publish(event_name, *args, error_strategy=None, **kwargs)

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

Parameters:

Name	Type	Description	Default
event_name	str	Имя события.	required
*args	Any	Позиционные аргументы.	()
error_strategy	ErrorStrategy | None	Стратегия обработки ошибок.	None
**kwargs	Any	Именованные аргументы.	{}

publish_async(event_name, *args, error_strategy=None, **kwargs) async

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

Parameters:

Name	Type	Description	Default
event_name	str	Имя события.	required
*args	Any	Позиционные аргументы.	()
error_strategy	ErrorStrategy | None	Стратегия обработки ошибок.	None
**kwargs	Any	Именованные аргументы.	{}

subscribe(event_name)

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

Parameters:

Name	Type	Description	Default
event_name	str	Имя события.	required

Returns:

Type	Description
Callable[[Callable[..., Any]], Callable[..., Any]]	Декоратор для функции-обработчика.

options: members:

• EventBus
• ErrorStrategy
• subscribe
• publish
• publish_async

Модуль tasks (Планировщик фоновых задач)

chutils.tasks

Модуль планировщика фоновых задач.

options: members:

• periodic_task
• start_scheduler
• stop_scheduler
• ErrorStrategy

Модуль di (Внедрение зависимостей)

chutils.di

options: members:

• Container
• provide
• inject
• Inject

Модуль metrics (Абстракция для метрик)

chutils.metrics

PROMETHEUS_AVAILABLE = importlib.util.find_spec('prometheus_client') is not None module-attribute

Глобальный флаг доступности prometheus_client

InMemoryMetricsProvider

Bases: MetricsProvider

Потокобезопасный in-memory провайдер метрик.

Не требует внешних зависимостей. Форматирует экспорт в стандартный текстовый формат Prometheus для бесшовной интеграции.

__init__()

Инициализирует InMemoryMetricsProvider.

clear()

Очищает все накопленные данные метрик.

generate_latest()

Экспортировать накопленные метрики в текстовом формате.

Returns:

Type	Description
str	Строка с отформатированными метриками.

get_metrics()

Возвращает сырые накопленные метрики в виде словаря (для отладки и тестов).

Returns:

Type	Description
dict[str, Any]	Словарь с сырыми данными по счетчикам, датчикам и гистограммам.

increment(name, value=1.0, labels=None)

Увеличить счетчик (Counter) на заданное значение.

Parameters:

Name	Type	Description	Default
name	str	Имя метрики.	required
value	float	Добавляемое значение.	1.0
labels	dict[str, str] | None	Словарь меток.	None

observe(name, value, labels=None)

Записать значение в гистограмму/таймер (Histogram/Timer).

Parameters:

Name	Type	Description	Default
name	str	Имя метрики.	required
value	float	Записываемое значение.	required
labels	dict[str, str] | None	Словарь меток.	None

set_gauge(name, value, labels=None)

Установить значение датчика (Gauge).

Parameters:

Name	Type	Description	Default
name	str	Имя датчика.	required
value	float	Устанавливаемое значение.	required
labels	dict[str, str] | None	Словарь меток.	None

MetricsProvider

Bases: ABC

Абстрактный базовый класс (интерфейс) для провайдеров метрик.

clear() abstractmethod

Очистить все накопленные данные (для тестов).

generate_latest() abstractmethod

Экспортировать накопленные метрики в текстовом формате.

Returns:

Type	Description
str	Строка с накопленными метриками.

increment(name, value=1.0, labels=None) abstractmethod

Увеличить счетчик (Counter) на заданное значение.

Parameters:

Name	Type	Description	Default
name	str	Имя метрики.	required
value	float	Значение, на которое нужно увеличить счетчик.	1.0
labels	dict[str, str] | None	Словарь меток для метрики.	None

observe(name, value, labels=None) abstractmethod

Записать значение в гистограмму/таймер (Histogram/Timer).

Parameters:

Name	Type	Description	Default
name	str	Имя метрики гистограммы/таймера.	required
value	float	Наблюдаемое значение.	required
labels	dict[str, str] | None	Словарь меток для метрики.	None

set_gauge(name, value, labels=None) abstractmethod

Установить значение датчика (Gauge).

Parameters:

Name	Type	Description	Default
name	str	Имя датчика.	required
value	float	Устанавливаемое значение датчика.	required
labels	dict[str, str] | None	Словарь меток для метрики.	None

PrometheusMetricsProvider

Bases: MetricsProvider

Провайдер метрик, транслирующий вызовы в prometheus_client.

Использует ленивый импорт. Если библиотека prometheus_client отсутствует, выбрасывает OptionalDependencyError при инициализации.

__init__()

Инициализирует PrometheusMetricsProvider.

clear()

Очистить кэш провайдера (но не глобальный REGISTRY Prometheus).

generate_latest()

Экспортировать накопленные метрики в текстовом формате.

Returns:

Type	Description
str	Строка с отформатированными метриками.

increment(name, value=1.0, labels=None)

Увеличить счетчик (Counter) на заданное значение.

Parameters:

Name	Type	Description	Default
name	str	Имя метрики.	required
value	float	Добавляемое значение.	1.0
labels	dict[str, str] | None	Словарь меток.	None

observe(name, value, labels=None)

Записать значение в гистограмму/таймер (Histogram/Timer).

Parameters:

Name	Type	Description	Default
name	str	Имя метрики.	required
value	float	Записываемое значение.	required
labels	dict[str, str] | None	Словарь меток.	None

set_gauge(name, value, labels=None)

Установить значение датчика (Gauge).

Parameters:

Name	Type	Description	Default
name	str	Имя датчика.	required
value	float	Устанавливаемое значение.	required
labels	dict[str, str] | None	Словарь меток.	None

TimerContext

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

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

with timer("db_query_duration_seconds", labels={"op": "select"}): db.execute("SELECT ...")

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

@timer("http_request_duration_seconds", labels={"endpoint": "/users"}) def handle(): ...

__call__(func)

Позволяет использовать TimerContext в качестве декоратора функции.

Parameters:

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

Returns:

Type	Description
F	Обернутая функция, замеряющая время своего выполнения.

__init__(name, labels=None)

Инициализирует TimerContext.

Parameters:

Name	Type	Description	Default
name	str	Имя метрики гистограммы для записи замера.	required
labels	dict[str, str] | None	Опциональные метки для метрики.	None

clear()

Очистить данные активного провайдера метрик.

generate_latest()

Сгенерировать дамп последних метрик в текстовом формате.

Returns:

Type	Description
str	Дамп метрик в формате Prometheus или пустая строка при ошибке.

get_provider()

Получить текущий активный провайдер метрик.

Если провайдер не задан вручную, инициализирует PrometheusMetricsProvider (если библиотека доступна) или InMemoryMetricsProvider в качестве fallback.

Returns:

Type	Description
MetricsProvider	Текущий активный экземпляр MetricsProvider.

increment(name, value=1.0, labels=None)

Увеличить счетчик (Counter) на заданное значение.

Parameters:

Name	Type	Description	Default
name	str	Имя метрики.	required
value	float	Значение, на которое нужно увеличить счетчик.	1.0
labels	dict[str, str] | None	Словарь меток для метрики.	None

observe(name, value, labels=None)

Записать значение в гистограмму/таймер (Histogram/Timer).

Parameters:

Name	Type	Description	Default
name	str	Имя метрики гистограммы/таймера.	required
value	float	Наблюдаемое значение.	required
labels	dict[str, str] | None	Словарь меток для метрики.	None

set_gauge(name, value, labels=None)

Установить значение датчика (Gauge).

Parameters:

Name	Type	Description	Default
name	str	Имя датчика.	required
value	float	Устанавливаемое значение датчика.	required
labels	dict[str, str] | None	Словарь меток для метрики.	None

set_provider(provider)

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

Parameters:

Name	Type	Description	Default
provider	MetricsProvider	Экземпляр MetricsProvider для установки.	required

options: members:

• increment
• set_gauge
• observe
• timer
• generate_latest
• get_provider
• set_provider
• clear

Исключения

chutils.exceptions

AuditError

Bases: ChutilsException

Базовый класс ошибок модуля audit.

AuditIntegrityError

Bases: AuditError

Ошибка целостности журнала аудита.

Выбрасывается при обнаружении нарушения криптографической цепочки хэшей.

BulkheadLimitExceeded

Bases: ChutilsException

Ошибка: превышен предел параллельных запросов Bulkhead.

CacheError

Bases: ChutilsException

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

ChutilsConfigurationError

Bases: ChutilsException

Ошибка конфигурации компонентов chutils.

ChutilsException

Bases: Exception

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

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

message property

Сообщение об ошибке.

Returns:

Type	Description
str	Текст сообщения об ошибке.

__init__(message, hint=None, **context)

Инициализирует базовое исключение ChutilsException.

Parameters:

Name	Type	Description	Default
message	str	Сообщение об ошибке.	required
hint	str | None	Опциональная подсказка по устранению ошибки.	None
**context	Any	Дополнительный контекст ошибки.	{}

ChutilsTimeoutError

Bases: ChutilsException

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

ChutilsValidationError

Bases: ChutilsException

Исключение при ошибке валидации данных.

__init__(message, errors=None, raw_error=None, hint=None, **context)

Инициализирует исключение валидации.

Parameters:

Name	Type	Description	Default
message	str	Сообщение об ошибке.	required
errors	list[dict[str, Any]] | None	Список ошибок в структурированном виде.	None
raw_error	Exception | None	Исходное исключение (например, ValidationError), если доступно.	None
hint	str | None	Опциональная подсказка.	None
**context	Any	Дополнительный контекст.	{}

__rich__()

Рендерит красивую таблицу ошибок валидации для rich.

Returns:

Type	Description
Any	Экземпляр rich.table.Table.

CircuitBreakerOpenError

Bases: ChutilsException

Ошибка: цепь предохранителя открыта (запросы заблокированы).

CommandError

Bases: ChutilsException

Ошибка при выполнении CLI команды.

ConfigError

Bases: ChutilsException

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

ConfigKeyNotFoundError

Bases: ConfigError

Ошибка: ключ или секция конфигурации не найдены.

ConfigLoadError

Bases: ConfigError

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

ConfigParseError

Bases: ConfigError

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

ConfigValidationGroupError

Bases: _BaseExceptionGroup, ConfigError

Группа ошибок валидации ключей конфигурации (отсутствие обязательных ключей).

__init__(message, exceptions, **context)

Инициализирует группу ошибок валидации конфигурации.

Parameters:

Name	Type	Description	Default
message	str	Сообщение об ошибке.	required
exceptions	list[Exception]	Список исключений ConfigKeyNotFoundError.	required
**context	Any	Дополнительный контекст ошибки.	{}

DependencyError

Bases: ChutilsException

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

DependencyNotFoundError

Bases: DependencyError

Ошибка: запрашиваемая зависимость не зарегистрирована в контейнере.

DependencyResolutionError

Bases: DependencyError

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

EnvValidationError

Bases: ChutilsException

Исключение при ошибке валидации переменных окружения.

__init__(message, errors=None, hint=None, **context)

Инициализирует исключение валидации переменных окружения.

Parameters:

Name	Type	Description	Default
message	str	Сообщение об ошибке.	required
errors	list[dict[str, Any]] | None	Список ошибок в структурированном виде.	None
hint	str | None	Опциональная подсказка.	None
**context	Any	Дополнительный контекст.	{}

__rich__()

Рендерит красивую таблицу ошибок для rich.

Returns:

Type	Description
Any	Экземпляр rich.table.Table.

EventBusError

Bases: ChutilsException

Общая ошибка шины событий.

EventBusExceptionGroup

Bases: _BaseExceptionGroup, EventBusError

Группа ошибок, возникших при параллельном или последовательном выполнении обработчиков событий шины.

__init__(message, exceptions, **context)

Инициализирует группу исключений шины событий.

Parameters:

Name	Type	Description	Default
message	str	Сообщение об ошибке.	required
exceptions	list[Exception]	Список перехваченных исключений от обработчиков.	required
**context	Any	Дополнительный контекст ошибки.	{}

FileSystemError

Bases: ChutilsException

Общая ошибка при работе с файловой системой.

HttpClientError

Bases: ChutilsException

Базовая ошибка HTTP-клиента chutils.

LoggerConfigurationError

Bases: ChutilsException

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

OptionalDependencyError

Bases: ChutilsException

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

PathTraversalError

Bases: FileSystemError

Ошибка безопасности: попытка выхода за пределы базовой директории (Path Traversal).

__init__(message, attempted_path='unknown', base_path='unknown', hint='Проверьте правильность пути или права доступа.', **context)

Инициализирует исключение попытки выхода за пределы базовой директории.

Parameters:

Name	Type	Description	Default
message	str	Сообщение об ошибке.	required
attempted_path	str | Path	Недопустимый путь, к которому пытались получить доступ.	'unknown'
base_path	str | Path	Базовый разрешенный путь.	'unknown'
hint	str | None	Опциональная подсказка для пользователя.	'Проверьте правильность пути или права доступа.'
**context	Any	Дополнительный контекст ошибки.	{}

RateLimitExceededError

Bases: ChutilsException

Ошибка: превышен лимит частоты вызовов (Rate Limit Exceeded).

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.

Модуль dev (AI-валидация и аудит)

chutils.dev

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

CleanItem dataclass

Элемент, предназначенный для очистки.

display_size property

Возвращает человекочитаемый размер элемента.

LintResult

Представляет результат одной проверки правила (Fallback версия без Pydantic).

__init__(rule_name, message, severity, file_path=None, line_number=None, fix_suggestion=None)

Инициализирует fallback-результат проверки правила.

Parameters:

Name	Type	Description	Default
rule_name	str	Название правила.	required
message	str	Сообщение об ошибке/предупреждении.	required
severity	str	Критичность проблемы.	required
file_path	str | None	Опциональный путь к файлу.	None
line_number	int | None	Номер строки.	None
fix_suggestion	str | None	Рекомендация по исправлению.	None

LinterEngine

Движок линтера, координирующий сбор файлов, загрузку правил и их выполнение.

__init__(config)

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

Parameters:

Name	Type	Description	Default
config	dict[str, str | bool | list[str] | None]	Словарь настроек линтера.	required

collect_all_files()

Собирает абсолютно все неигнорируемые файлы в проекте.

Returns:

Type	Description
list[str]	Список абсолютных путей к файлам.

collect_files()

Собирает все неигнорируемые файлы в проекте (учитывая флаг staged).

Returns:

Type	Description
list[str]	Список абсолютных путей к файлам.

collect_staged_files()

Собирает список измененных и добавленных файлов, подготовленных к коммиту (staged) в Git.

Returns:

Type	Description
list[str]	Список абсолютных путей к файлам.

load_rules()

Загружает правила (встроенные и кастомные).

print_results(results)

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

Parameters:

Name	Type	Description	Default
results	list[LintResult]	Список результатов.	required

Returns:

Type	Description
bool	True, если проверка успешна (нет критических ошибок в строгом режиме/обычном),
bool	False в противном случае.

run()

Запускает все включенные правила на собранных файлах.

Returns:

Type	Description
list[LintResult]	Список результатов проверок с найденными ошибками и предупреждениями.

should_ignore(path)

Проверяет, должен ли данный путь быть проигнорирован.

Parameters:

Name	Type	Description	Default
path	Path	Проверяемый путь.	required

Returns:

Type	Description
bool	True, если путь соответствует какому-либо шаблону игнорирования.

MockServerRunner

Управляющий класс для мок-сервера.

__init__(port=8888, routes_path='mocks.yml', proxy_fallback=None)

Инициализирует MockServerRunner.

Parameters:

Name	Type	Description	Default
port	int	Порт, на котором будет запущен мок-сервер.	8888
routes_path	str	Путь к YAML-файлу с описанием роутов.	'mocks.yml'
proxy_fallback	str | None	URL реального бэкенда для проксирования неизвестных роутов.	None

check_reload()

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

init_template(output_path)

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

Parameters:

Name	Type	Description	Default
output_path	str	Путь к файлу для сохранения шаблона.	required

load_config()

Загружает роуты из файла (YAML или JSON).

log_event(message)

Выводит отформатированное лог-сообщение в консоль.

Parameters:

Name	Type	Description	Default
message	str	Текст лог-сообщения.	required

run()

Запускает многопоточный HTTP-сервер.

stop()

Останавливает запущенный HTTP-сервер.

Rule

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

Каждое правило должно переопределить метод check() и задать атрибуты name, description и severity.

Подавление срабатываний инлайн (inline suppress)

Любое срабатывание правила можно подавить без изменения правил, добавив комментарий в конец проблемной строки или в строку непосредственно перед ней::

import logging  # chutils: ignore[ChutilsIntegrationRule]
code()         # chutils: ignore[RuleA, RuleB]
# chutils: ignore[ChutilsIntegrationRule]
some_call()

Для подавления всех правил сразу используйте all::

code()  # chutils: ignore[all]

Правила не должны сами проверять инлайн-комментарии: фильтрация выполняется автоматически в LinterEngine.run().

check(base_dir, files)

Выполняет проверку правила по списку файлов.

Parameters:

Name	Type	Description	Default
base_dir	str	Путь к корню проверяемого проекта.	required
files	list[str]	Список абсолютных путей к файлам проекта.	required

Returns:

Type	Description
list[LintResult]	Список объектов LintResult с найденными проблемами.

Note

Подавить срабатывания отдельного срабатывания можно инлайн-комментарием # chutils: ignore[<name>] — без изменения правила.

collect_context_slice(project_path, modules=None, task=None, layer='public')

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

Parameters:

Name	Type	Description	Default
project_path	Path	Путь к корню проекта.	required
modules	list[str] | None	Список выбранных модулей.	None
task	str | None	Описание задачи.	None
layer	str	Слой абстракции для фильтрации символов.	'public'

Returns:

Type	Description
str	Сгенерированный Markdown с контекстом.

execute_clean(items)

Удаляет найденные мусорные элементы.

Parameters:

Name	Type	Description	Default
items	list[CleanItem]	Список элементов CleanItem для удаления.	required

Returns:

Type	Description
tuple[int, int]	Кортеж (количество удаленных элементов, суммарно освобожденный размер в байтах).

generate_few_shot_bank(project_path, *, force=False, console=None)

Генерирует банк few-shot примеров для целевого проекта.

Анализирует архитектуру проекта, детектирует ключевые абстракции (Use Cases, репозитории, логгеры, исключения, DI-контейнеры) и создаёт параметризованные шаблоны docs/ai_examples/ в корне целевого проекта.

Parameters:

Name	Type	Description	Default
project_path	str	Путь к корневой директории целевого проекта.	required
force	bool	Если True, существующие категории будут перезаписаны.	False
console	_ConsoleProtocol | None	Объект Rich Console для вывода статуса (опционально).	None

Returns:

Type	Description
GenerationResult	GenerationResult с детализацией созданных и пропущенных категорий.

Raises:

Type	Description
FileNotFoundError	Если project_path не существует.
ValueError	При нарушении path traversal защиты.

generate_workflow_yaml(python_versions, with_pytest, with_mypy, with_ruff, with_ai_lint)

Генерирует валидный YAML-конфиг для GitHub Actions на основе setup-uv.

Parameters:

Name	Type	Description	Default
python_versions	list[str]	Список версий Python для матрицы тестирования.	required
with_pytest	bool	Запускать ли тесты с pytest.	required
with_mypy	bool	Запускать ли статический анализ типов с mypy.	required
with_ruff	bool	Запускать ли линтинг кода с ruff.	required
with_ai_lint	bool	Запускать ли аудит готовности к AI с chutils dev ai-lint.	required

Returns:

Type	Description
str	Строка с содержимым YAML-файла.

run_interactive_menu(project_path)

Запускает красивое интерактивное CLI-меню для выбора модулей.

Parameters:

Name	Type	Description	Default
project_path	Path	Путь к корню проекта.	required

Returns:

Type	Description
list[str]	Список выбранных пользователем модулей.

scan_project(base_dir, excludes=None, default_targets=None, extra_targets=None)

Сканирует проект и возвращает список найденных временных файлов и папок.

Parameters:

Name	Type	Description	Default
base_dir	str | Path	Корневая директория проекта.	required
excludes	list[str] | None	Список папок или шаблонов для исключения из обхода.	None
default_targets	list[str] | None	Базовый список шаблонов временных файлов/папок.	None
extra_targets	list[str] | None	Дополнительные шаблоны для очистки.	None

Returns:

Type	Description
list[CleanItem]	Список объектов CleanItem.

options: members:

• Rule
• LintResult
• LinterEngine
• collect_context_slice
• run_interactive_menu
• generate_few_shot_bank
• MockServerRunner
• Scaffolder
• generate_workflow_yaml

Модуль diagnostics (Мониторинг работоспособности)

chutils.diagnostics

CheckResult dataclass

Результат выполнения проверки диагностики (вариант без Pydantic).

Attributes:

Name	Type	Description
name	str	Название проверки.
success	bool	Флаг успешности проверки.
critical	bool	Флаг критичности проверки.
execution_time	float	Время выполнения проверки в секундах.
error	str | None	Текст ошибки, если проверка завершилась неудачно.
message	str | None	Дополнительное информационное сообщение.

model_dump()

Преобразует модель в словарь.

Returns:

Type	Description
dict[str, str | bool | float | None]	Словарь с данными о результате проверки.

HealthReport dataclass

Отчет о состоянии работоспособности системы (вариант без Pydantic).

Attributes:

Name	Type	Description
status	str	Общий статус системы (HEALTHY, DEGRADED, UNHEALTHY).
results	list[CheckResult]	Список результатов проверок.
total_time	float	Общее время выполнения всех проверок в секундах.

model_dump()

Преобразует отчет в словарь.

Returns:

Type	Description
dict[str, str | list[dict[str, str | bool | float | None]] | float]	Словарь с данными о состоянии здоровья системы.

options: members:

• DiagnosticsManager
• CheckResult
• HealthReport
• get_fastapi_health_handler
• get_flask_health_handler

Модуль env (Манифест окружения)

chutils.env

BaseEnvManifest

Заглушка манифеста переменных окружения (Pydantic не установлен).

load() classmethod

Пытается загрузить манифест без Pydantic.

Returns:

Type	Description
T	Метод никогда не возвращает значение, так как всегда вызывает исключение.

Raises:

Type	Description
OptionalDependencyError	Всегда выбрасывается, так как Pydantic отсутствует.

has_pydantic()

Возвращает True, если Pydantic установлен.

Returns:

Type	Description
bool	True, если пакет pydantic доступен для импорта.

has_rich()

Возвращает True, если Rich установлен.

Returns:

Type	Description
bool	True, если пакет rich доступен для импорта.

has_watchdog()

Возвращает True, если Watchdog установлен.

Returns:

Type	Description
bool	True, если пакет watchdog доступен для импорта.

is_otel_enabled()

Проверяет, доступен ли OpenTelemetry и не отключен ли он.

Учитывает: - Наличие установленного пакета opentelemetry. - Переменную окружения CH_DISABLE_TRACING

Returns:

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

is_rich_enabled()

Централизованная проверка: доступен ли Rich и разрешен ли он настройками.

Учитывает: - Наличие установленного пакета rich. - Переменные окружения NO_COLOR, CH_NO_COLOR. - Специальную переменную CH_NO_RICH (для тестов и headless).

Returns:

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

options: members:

• BaseEnvManifest
• is_rich_enabled
• is_otel_enabled

Модуль validation (Валидация данных)

chutils.validation

validate_call(func)

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

Использует pydantic.validate_call под капотом, если pydantic установлен. В случае ошибки валидации выбрасывает ChutilsValidationError.

Parameters:

Name	Type	Description	Default
func	Callable[P, R]	Декорируемая функция (синхронная или асинхронная).	required

Returns:

Type	Description
Callable[P, R]	Декорированная функция с автоматической валидацией аргументов.

Raises:

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

validate_data(model, data)

Валидирует словарь или JSON-строку по заданной Pydantic модели.

Parameters:

Name	Type	Description	Default
model	type[T]	Класс Pydantic модели для валидации.	required
data	dict[str, Any] | str	Данные для валидации в виде словаря или JSON-строки.	required

Returns:

Type	Description
T	Экземпляр провалидированной модели Pydantic.

Raises:

Type	Description
OptionalDependencyError	Если пакет pydantic не установлен.
ChutilsValidationError	Если данные не прошли валидацию или JSON невалиден.

options: members:

• validate_data
• validate_call

Модуль web (Умный HTTP-клиент)

chutils.web

AsyncWebClient

Bases: AsyncClient

Асинхронный HTTP-клиент с поддержкой ротации User-Agent, прокси-пулов,

лимитирования частоты и кэширования GET-запросов.

__init__(*args, user_agent_rotator=None, proxy_pool=None, rotate_ua=True, rotate_proxy=True, retries=0, retry_delay=1.0, retry_backoff=2.0, retry_on_5xx=True, rate_limit_calls=None, rate_limit_period=1.0, rate_limit_strategy='token_bucket', rate_limit_wait=False, cache_ttl=None, cache_backend=None, **kwargs)

Инициализирует AsyncWebClient.

Parameters:

Name	Type	Description	Default
*args	Any	Позиционные аргументы для родительского класса httpx.AsyncClient.	()
user_agent_rotator	UserAgentRotator | None	Ротатор User-Agent.	None
proxy_pool	ProxyPool | None	Менеджер пула прокси.	None
rotate_ua	bool	Включить ли ротацию User-Agent.	True
rotate_proxy	bool	Включить ли смену прокси при ошибках.	True
retries	int	Количество повторов при ошибках.	0
retry_delay	float	Базовая задержка между попытками в секундах.	1.0
retry_backoff	float	Множитель задержки.	2.0
retry_on_5xx	bool	Считать ли 5xx ошибки сбоем для повтора.	True
rate_limit_calls	int | None	Ограничение количества запросов к хосту.	None
rate_limit_period	float	Период ограничения в секундах.	1.0
rate_limit_strategy	str	Стратегия лимитирования ('token_bucket' или 'leaky_bucket').	'token_bucket'
rate_limit_wait	bool	Ждать ли освобождения токена (или вызывать исключение).	False
cache_ttl	int | None	Время жизни кэша GET-запросов в секундах.	None
cache_backend	Any | None	Бэкенд кэша.	None
**kwargs	Any	Именованные аргументы для родительского класса httpx.AsyncClient.	{}

rotate_proxy() async

Асинхронно переключает текущий клиент на следующий прокси из пула.

send(request, *args, **kwargs) async

Перехватывает отправку запроса для ротации, лимитов и кэширования.

Parameters:

Name	Type	Description	Default
request	Request	Объект отправляемого запроса httpx.Request.	required
*args	Any	Произвольные позиционные аргументы.	()
**kwargs	Any	Произвольные именованные аргументы.	{}

Returns:

Type	Description
Response	Ответ сервера httpx.Response.

WebClient

Bases: Client

Синхронный HTTP-клиент с поддержкой ротации User-Agent, прокси-пулов,

лимитирования частоты и кэширования GET-запросов.

__init__(*args, user_agent_rotator=None, proxy_pool=None, rotate_ua=True, rotate_proxy=True, retries=0, retry_delay=1.0, retry_backoff=2.0, retry_on_5xx=True, rate_limit_calls=None, rate_limit_period=1.0, rate_limit_strategy='token_bucket', rate_limit_wait=False, cache_ttl=None, cache_backend=None, **kwargs)

Инициализирует WebClient.

Parameters:

Name	Type	Description	Default
*args	Any	Позиционные аргументы для родительского класса httpx.Client.	()
user_agent_rotator	UserAgentRotator | None	Ротатор User-Agent.	None
proxy_pool	ProxyPool | None	Менеджер пула прокси.	None
rotate_ua	bool	Включить ли ротацию User-Agent.	True
rotate_proxy	bool	Включить ли смену прокси при ошибках.	True
retries	int	Количество повторов при ошибках.	0
retry_delay	float	Базовая задержка между попытками в секундах.	1.0
retry_backoff	float	Множитель задержки.	2.0
retry_on_5xx	bool	Считать ли 5xx ошибки сбоем для повтора.	True
rate_limit_calls	int | None	Ограничение количества запросов к хосту.	None
rate_limit_period	float	Период ограничения в секундах.	1.0
rate_limit_strategy	str	Стратегия лимитирования ('token_bucket' или 'leaky_bucket').	'token_bucket'
rate_limit_wait	bool	Ждать ли освобождения токена (или вызывать исключение).	False
cache_ttl	int | None	Время жизни кэша GET-запросов в секундах.	None
cache_backend	Any | None	Бэкенд кэша.	None
**kwargs	Any	Именованные аргументы для родительского класса httpx.Client.	{}

rotate_proxy()

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

send(request, *args, **kwargs)

Перехватывает отправку запроса для ротации, лимитов и кэширования.

Parameters:

Name	Type	Description	Default
request	Request	Объект отправляемого запроса httpx.Request.	required
*args	Any	Произвольные позиционные аргументы.	()
**kwargs	Any	Произвольные именованные аргументы.	{}

Returns:

Type	Description
Response	Ответ сервера httpx.Response.

options: members:

• WebClient
• AsyncWebClient

Модуль scraping.humanize (Имитация поведения человека)

chutils.scraping.humanize

BezierCurveGenerator

Генератор траекторий перемещения на основе кривых Безье.

generate(start, end, steps=30, deviation=0.2)

Генерирует сглаженную траекторию от start к end.

Parameters:

Name	Type	Description	Default
start	tuple[int, int]	Начальная координата (x, y).	required
end	tuple[int, int]	Конечная координата (x, y).	required
steps	int	Количество шагов (точек) в траектории.	30
deviation	float	Максимальное отклонение контрольных точек от прямой линии.	0.2

Returns:

Type	Description
list[tuple[int, int]]	Список координат точек траектории движения.

JitterDelayGenerator

Генератор реалистичных задержек.

__init__(strategy='lognormal', jitter=0.15)

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

Parameters:

Name	Type	Description	Default
strategy	str	Стратегия ('lognormal' или 'normal').	'lognormal'
jitter	float	Коэффициент разброса (джиттер).	0.15

generate(base_delay)

Возвращает сгенерированную задержку на основе базовой.

Parameters:

Name	Type	Description	Default
base_delay	float	Базовая величина задержки (в секундах).	required

Returns:

Type	Description
float	Полученное случайное значение задержки с учетом джиттера.

KeyboardTypoGenerator

Генератор последовательностей ввода символов с реалистичными опечатками.

generate_sequence(text, error_rate=0.05)

Генерирует последовательность нажатий клавиш для ввода текста.

Включает случайные опечатки, их обнаружение и исправление через Backspace.

Parameters:

Name	Type	Description	Default
text	str	Исходный текст.	required
error_rate	float	Вероятность совершения ошибки на каждом символе.	0.05

Returns:

Type	Description
list[TypoAction]	Список действий TypoAction, имитирующий последовательный ввод текста человеком.

ProfileWarmer

Класс для асинхронного прогрева браузерных профилей (Playwright, nodriver).

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

__init__(browser_or_tab)

Инициализирует ProfileWarmer.

Parameters:

Name	Type	Description	Default
browser_or_tab	Any	Объект страницы Playwright Page или вкладки nodriver Tab.	required

warm_up(sites=None, sites_count=3, duration_per_site=(10.0, 30.0), click_random_links=True) async

Запускает процесс прогрева профиля.

Parameters:

Name	Type	Description	Default
sites	list[str] | None	Список URL-адресов трастовых сайтов для прогрева. Если None, используется встроенный список.	None
sites_count	int	Количество посещаемых сайтов.	3
duration_per_site	tuple[float, float]	Диапазон времени пребывания на одном сайте (мин, макс в секундах).	(10.0, 30.0)
click_random_links	bool	Флаг перехода по случайным внутренним ссылкам.	True

SyncProfileWarmer

Класс для синхронного прогрева браузерных профилей (Selenium).

__init__(driver)

Инициализирует SyncProfileWarmer.

Parameters:

Name	Type	Description	Default
driver	Any	Экземпляр Selenium WebDriver.	required

warm_up(sites=None, sites_count=3, duration_per_site=(10.0, 30.0), click_random_links=True)

Запускает процесс прогрева профиля (синхронно).

Parameters:

Name	Type	Description	Default
sites	list[str] | None	Список URL-адресов трастовых сайтов для прогрева. Если None, используется встроенный список.	None
sites_count	int	Количество посещаемых сайтов.	3
duration_per_site	tuple[float, float]	Диапазон времени пребывания на одном сайте (мин, макс в секундах).	(10.0, 30.0)
click_random_links	bool	Флаг перехода по случайным внутренним ссылкам.	True

apply_antidetect_nodriver(tab, *, webgl_vendor=DEFAULT_WEBGL_VENDOR, webgl_renderer=DEFAULT_WEBGL_RENDERER, hardware_concurrency=DEFAULT_HARDWARE_CONCURRENCY, device_memory=DEFAULT_DEVICE_MEMORY) async

Применяет JS-инъекции анти-детекта к вкладке (Tab) nodriver.

Parameters:

Name	Type	Description	Default
tab	Any	Объект вкладки nodriver Tab.	required
webgl_vendor	str	Подменяемый производитель WebGL.	DEFAULT_WEBGL_VENDOR
webgl_renderer	str	Подменяемая видеокарта WebGL.	DEFAULT_WEBGL_RENDERER
hardware_concurrency	int	Эмулируемое количество ядер процессора.	DEFAULT_HARDWARE_CONCURRENCY
device_memory	int	Эмулируемый объем оперативной памяти в ГБ.	DEFAULT_DEVICE_MEMORY

apply_antidetect_playwright(context, *, webgl_vendor=DEFAULT_WEBGL_VENDOR, webgl_renderer=DEFAULT_WEBGL_RENDERER, hardware_concurrency=DEFAULT_HARDWARE_CONCURRENCY, device_memory=DEFAULT_DEVICE_MEMORY) async

Применяет JS-инъекции анти-детекта к контексту Playwright.

Parameters:

Name	Type	Description	Default
context	Any	Объект контекста Playwright BrowserContext.	required
webgl_vendor	str	Подменяемый производитель WebGL.	DEFAULT_WEBGL_VENDOR
webgl_renderer	str	Подменяемая видеокарта WebGL.	DEFAULT_WEBGL_RENDERER
hardware_concurrency	int	Эмулируемое количество ядер процессора.	DEFAULT_HARDWARE_CONCURRENCY
device_memory	int	Эмулируемый объем оперативной памяти в ГБ.	DEFAULT_DEVICE_MEMORY

apply_antidetect_selenium(driver, *, webgl_vendor=DEFAULT_WEBGL_VENDOR, webgl_renderer=DEFAULT_WEBGL_RENDERER, hardware_concurrency=DEFAULT_HARDWARE_CONCURRENCY, device_memory=DEFAULT_DEVICE_MEMORY)

Применяет JS-инъекции анти-детекта к сессии Selenium.

Parameters:

Name	Type	Description	Default
driver	Any	Экземпляр Selenium WebDriver.	required
webgl_vendor	str	Подменяемый производитель WebGL.	DEFAULT_WEBGL_VENDOR
webgl_renderer	str	Подменяемая видеокарта WebGL.	DEFAULT_WEBGL_RENDERER
hardware_concurrency	int	Эмулируемое количество ядер процессора.	DEFAULT_HARDWARE_CONCURRENCY
device_memory	int	Эмулируемый объем оперативной памяти в ГБ.	DEFAULT_DEVICE_MEMORY

async_human_sleep(min_seconds, max_seconds) async

Асинхронно задерживает выполнение на случайное время, имитируя поведение человека.

Parameters:

Name	Type	Description	Default
min_seconds	float	Минимальное время задержки (в секундах).	required
max_seconds	float	Максимальное время задержки (в секундах).	required

async_move_mouse(page, x, y, start=None, steps=30, delay_between_steps=0.01) async

Имитирует плавное перемещение мыши Playwright или nodriver.

Parameters:

Name	Type	Description	Default
page	Any	Объект страницы Playwright Page или вкладки nodriver Tab.	required
x	int	Конечная координата X.	required
y	int	Конечная координата Y.	required
start	tuple[int, int] | None	Начальные координаты X, Y. Если не задано, используется (0, 0).	None
steps	int	Количество промежуточных шагов движения.	30
delay_between_steps	float	Задержка между шагами в секундах.	0.01

async_scroll_to(page, x, y, selector=None, steps=10, delay_between_steps=0.01) async

Имитирует плавный скроллинг Playwright или nodriver.

Parameters:

Name	Type	Description	Default
page	Any	Объект страницы Playwright Page или вкладки nodriver Tab.	required
x	int	Конечная горизонтальная позиция скролла.	required
y	int	Конечная вертикальная позиция скролла.	required
selector	str | None	Необязательный селектор элемента для скролла.	None
steps	int	Количество промежуточных шагов.	10
delay_between_steps	float	Задержка между шагами в секундах.	0.01

async_type_text(page, selector, text, error_rate=0.05, speed_wpm=40.0) async

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

Parameters:

Name	Type	Description	Default
page	Any	Объект страницы Playwright Page или вкладки nodriver Tab.	required
selector	str	Селектор поля ввода.	required
text	str	Текст для ввода.	required
error_rate	float	Вероятность совершения опечатки (0.0 - 1.0).	0.05
speed_wpm	float	Скорость ввода в словах в минуту (WPM).	40.0

get_browser_launch_args()

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

Returns:

Type	Description
list[str]	Список аргументов командной строки запуска браузера.

human_sleep(min_seconds, max_seconds)

Синхронно задерживает выполнение на случайное время, имитируя поведение человека.

Parameters:

Name	Type	Description	Default
min_seconds	float	Минимальное время задержки (в секундах).	required
max_seconds	float	Максимальное время задержки (в секундах).	required

move_mouse(driver, x, y, start=None, steps=30, delay_between_steps=0.01)

Имитирует плавное перемещение мыши Selenium.

Parameters:

Name	Type	Description	Default
driver	Any	Экземпляр Selenium WebDriver.	required
x	int	Конечная координата X.	required
y	int	Конечная координата Y.	required
start	tuple[int, int] | None	Начальные координаты X, Y. Если не задано, используется (0, 0).	None
steps	int	Количество промежуточных шагов.	30
delay_between_steps	float	Задержка между шагами в секундах.	0.01

scroll_to(driver, x, y, selector=None, steps=10, delay_between_steps=0.01)

Имитирует плавный скроллинг Selenium.

Parameters:

Name	Type	Description	Default
driver	Any	Экземпляр Selenium WebDriver.	required
x	int	Конечная горизонтальная позиция скролла.	required
y	int	Конечная вертикальная позиция скролла.	required
selector	str | None	Необязательный селектор элемента для скролла.	None
steps	int	Количество промежуточных шагов.	10
delay_between_steps	float	Задержка между шагами в секундах.	0.01

type_text(driver, selector, text, error_rate=0.05, speed_wpm=40.0)

Имитирует ввод текста с опечатками Selenium.

Parameters:

Name	Type	Description	Default
driver	Any	Экземпляр Selenium WebDriver.	required
selector	str	CSS-селектор поля ввода.	required
text	str	Текст для ввода.	required
error_rate	float	Вероятность совершения опечатки (0.0 - 1.0).	0.05
speed_wpm	float	Скорость ввода в словах в минуту (WPM).	40.0

options: members:

• BezierCurveGenerator
• JitterDelayGenerator
• KeyboardTypoGenerator
• human_sleep
• async_human_sleep
• move_mouse
• async_move_mouse
• scroll_to
• async_scroll_to
• type_text
• async_type_text
• apply_antidetect_playwright
• apply_antidetect_selenium
• get_browser_launch_args

Модуль scraping.captcha (Решатели капчи)

chutils.scraping.captcha

AntiCaptchaSolver

Bases: BaseCaptchaSolver

Синхронный клиент для Anti-Captcha.

__init__(api_key=None, host='https://api.anti-captcha.com')

Инициализирует AntiCaptchaSolver.

Parameters:

Name	Type	Description	Default
api_key	str | None	Явно заданный API-ключ.	None
host	str	Адрес API-сервиса Anti-Captcha.	'https://api.anti-captcha.com'

solve_image(image_data, timeout=60.0, poll_interval=5.0, **kwargs)

Синхронно решает капчу-изображение.

Parameters:

Name	Type	Description	Default
image_data	bytes | str	Бинарные данные картинки или base64-строка.	required
timeout	float	Максимальное время ожидания решения (в секундах).	60.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Распознанный текст с изображения.

solve_recaptcha(sitekey, page_url, timeout=120.0, poll_interval=5.0, **kwargs)

Синхронно решает ReCaptcha v2/v3.

Parameters:

Name	Type	Description	Default
sitekey	str	Ключ рекапчи на целевом сайте.	required
page_url	str	URL страницы, на которой расположена рекапча.	required
timeout	float	Максимальное время ожидания решения (в секундах).	120.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Токен ответа рекапчи.

AsyncAntiCaptchaSolver

Bases: BaseAsyncCaptchaSolver

Асинхронный клиент для Anti-Captcha.

__init__(api_key=None, host='https://api.anti-captcha.com')

Инициализирует AsyncAntiCaptchaSolver.

Parameters:

Name	Type	Description	Default
api_key	str | None	Явно заданный API-ключ.	None
host	str	Адрес API-сервиса Anti-Captcha.	'https://api.anti-captcha.com'

solve_image(image_data, timeout=60.0, poll_interval=5.0, **kwargs) async

Асинхронно решает капчу-изображение.

Parameters:

Name	Type	Description	Default
image_data	bytes | str	Бинарные данные картинки или base64-строка.	required
timeout	float	Максимальное время ожидания решения (в секундах).	60.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Распознанный текст с изображения.

solve_recaptcha(sitekey, page_url, timeout=120.0, poll_interval=5.0, **kwargs) async

Асинхронно решает ReCaptcha v2/v3.

Parameters:

Name	Type	Description	Default
sitekey	str	Ключ рекапчи на целевом сайте.	required
page_url	str	URL страницы, на которой расположена рекапча.	required
timeout	float	Максимальное время ожидания решения (в секундах).	120.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Токен ответа рекапчи.

AsyncCapMonsterSolver

Bases: BaseAsyncCaptchaSolver

Асинхронный клиент для CapMonster Cloud.

__init__(api_key=None, host='https://api.capmonster.cloud')

Инициализирует AsyncCapMonsterSolver.

Parameters:

Name	Type	Description	Default
api_key	str | None	Явно заданный API-ключ.	None
host	str	Адрес API-сервиса CapMonster Cloud.	'https://api.capmonster.cloud'

solve_image(image_data, timeout=60.0, poll_interval=5.0, **kwargs) async

Асинхронно решает капчу-изображение.

Parameters:

Name	Type	Description	Default
image_data	bytes | str	Бинарные данные картинки или base64-строка.	required
timeout	float	Максимальное время ожидания решения (в секундах).	60.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Распознанный текст с изображения.

solve_recaptcha(sitekey, page_url, timeout=120.0, poll_interval=5.0, **kwargs) async

Асинхронно решает ReCaptcha v2/v3.

Parameters:

Name	Type	Description	Default
sitekey	str	Ключ рекапчи на целевом сайте.	required
page_url	str	URL страницы, на которой расположена рекапча.	required
timeout	float	Максимальное время ожидания решения (в секундах).	120.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Токен ответа рекапчи.

AsyncRuCaptchaSolver

Bases: BaseAsyncCaptchaSolver

Асинхронный клиент для RuCaptcha / 2Captcha.

__init__(api_key=None, host='https://rucaptcha.com')

Инициализирует AsyncRuCaptchaSolver.

Parameters:

Name	Type	Description	Default
api_key	str | None	Явно заданный API-ключ.	None
host	str	Адрес API-сервиса RuCaptcha.	'https://rucaptcha.com'

solve_image(image_data, timeout=60.0, poll_interval=5.0, **kwargs) async

Асинхронно решает капчу-изображение.

Parameters:

Name	Type	Description	Default
image_data	bytes | str	Бинарные данные картинки или base64-строка.	required
timeout	float	Максимальное время ожидания решения (в секундах).	60.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Распознанный текст с изображения.

solve_recaptcha(sitekey, page_url, timeout=120.0, poll_interval=5.0, **kwargs) async

Асинхронно решает ReCaptcha v2/v3.

Parameters:

Name	Type	Description	Default
sitekey	str	Ключ рекапчи на целевом сайте.	required
page_url	str	URL страницы, на которой расположена рекапча.	required
timeout	float	Максимальное время ожидания решения (в секундах).	120.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Токен ответа рекапчи.

CapMonsterSolver

Bases: BaseCaptchaSolver

Синхронный клиент для CapMonster Cloud.

__init__(api_key=None, host='https://api.capmonster.cloud')

Инициализирует CapMonsterSolver.

Parameters:

Name	Type	Description	Default
api_key	str | None	Явно заданный API-ключ.	None
host	str	Адрес API-сервиса CapMonster Cloud.	'https://api.capmonster.cloud'

solve_image(image_data, timeout=60.0, poll_interval=5.0, **kwargs)

Синхронно решает капчу-изображение.

Parameters:

Name	Type	Description	Default
image_data	bytes | str	Бинарные данные картинки или base64-строка.	required
timeout	float	Максимальное время ожидания решения (в секундах).	60.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Распознанный текст с изображения.

solve_recaptcha(sitekey, page_url, timeout=120.0, poll_interval=5.0, **kwargs)

Синхронно решает ReCaptcha v2/v3.

Parameters:

Name	Type	Description	Default
sitekey	str	Ключ рекапчи на целевом сайте.	required
page_url	str	URL страницы, на которой расположена рекапча.	required
timeout	float	Максимальное время ожидания решения (в секундах).	120.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Токен ответа рекапчи.

CaptchaBalanceError

Bases: CaptchaError

Ошибка: недостаточный баланс на аккаунте сервиса капчи.

CaptchaError

Bases: ChutilsException

Базовая ошибка при решении капчи.

CaptchaServiceError

Bases: CaptchaError

Ошибка: сервис решения капчи вернул ошибку API (например, неверный ключ, плохие параметры).

CaptchaTimeoutError

Bases: CaptchaError

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

RuCaptchaSolver

Bases: BaseCaptchaSolver

Синхронный клиент для RuCaptcha / 2Captcha.

__init__(api_key=None, host='https://rucaptcha.com')

Инициализирует RuCaptchaSolver.

Parameters:

Name	Type	Description	Default
api_key	str | None	Явно заданный API-ключ.	None
host	str	Адрес API-сервиса RuCaptcha.	'https://rucaptcha.com'

solve_image(image_data, timeout=60.0, poll_interval=5.0, **kwargs)

Синхронно решает капчу-изображение.

Parameters:

Name	Type	Description	Default
image_data	bytes | str	Бинарные данные картинки или base64-строка.	required
timeout	float	Максимальное время ожидания решения (в секундах).	60.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Распознанный текст с изображения.

solve_recaptcha(sitekey, page_url, timeout=120.0, poll_interval=5.0, **kwargs)

Синхронно решает ReCaptcha v2/v3.

Parameters:

Name	Type	Description	Default
sitekey	str	Ключ рекапчи на целевом сайте.	required
page_url	str	URL страницы, на которой расположена рекапча.	required
timeout	float	Максимальное время ожидания решения (в секундах).	120.0
poll_interval	float	Интервал опроса статуса решения (в секундах).	5.0
**kwargs	Any	Дополнительные параметры задачи.	{}

Returns:

Type	Description
str	Токен ответа рекапчи.

options: members:

• RuCaptchaSolver
• AsyncRuCaptchaSolver
• AntiCaptchaSolver
• AsyncAntiCaptchaSolver
• CapMonsterSolver
• AsyncCapMonsterSolver
• CaptchaError
• CaptchaTimeoutError
• CaptchaBalanceError
• CaptchaServiceError

Модуль plugins (Система плагинов)

chutils.plugins

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

registry = PluginRegistry() module-attribute

Глобальный экземпляр реестра

BasePlugin

Bases: ABC

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

description property

Описание плагина.

Returns:

Type	Description
str	Описание плагина.

name abstractmethod property

Уникальное имя плагина.

Returns:

Type	Description
str	Имя плагина.

version property

Версия плагина.

Returns:

Type	Description
str	Строка с версией плагина.

ConfigProviderPlugin

Bases: BasePlugin, ConfigProvider

Интерфейс для плагина-провайдера конфигураций. Позволяет загружать и сохранять конфигурации из внешних систем (например, Consul, Etcd).

LoggerHandlerPlugin

Bases: BasePlugin

Интерфейс для плагина-обработчика логов. Позволяет добавлять кастомные logging.Handler в конфигурацию логирования.

get_handler(**kwargs) abstractmethod

Создает и возвращает настроенный экземпляр logging.Handler.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Параметры конфигурации для инициализации хэндлера.	{}

Returns:

Type	Description
Handler	Настроенный объект logging.Handler.

MetricsPlugin

Bases: BasePlugin, MetricsProvider

Интерфейс для плагина-провайдера метрик. Позволяет подключить стороннюю систему сбора метрик (например, Datadog, StatsD).

PluginError

Bases: Exception

Базовое исключение для ошибок системы плагинов.

PluginRegistry

Реестр плагинов chutils. Управляет жизненным циклом, регистрацией и автообнаружением плагинов.

__init__()

Инициализирует PluginRegistry.

clear()

Очистить все зарегистрированные плагины (в основном для тестов).

discover_plugins(group='chutils.plugins')

Автоматическое обнаружение плагинов через Python entry_points. Исключения при загрузке плагина логируются, но не прерывают работу всей системы.

Parameters:

Name	Type	Description	Default
group	str	Имя группы entry points для поиска плагинов.	'chutils.plugins'

get_all_plugins()

Получить список всех зарегистрированных плагинов.

Returns:

Type	Description
list[Any]	Список всех зарегистрированных плагинов.

get_plugin(name)

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

Parameters:

Name	Type	Description	Default
name	str	Имя плагина.	required

Returns:

Type	Description
Any | None	Экземпляр плагина или None, если он не найден.

get_plugins_by_type(plugin_type)

Получить все плагины, которые являются экземплярами или наследниками указанного типа.

Parameters:

Name	Type	Description	Default
plugin_type	type[Any]	Класс/тип плагина для фильтрации.	required

Returns:

Type	Description
list[Any]	Список плагинов, соответствующих указанному типу.

register(plugin)

Явно зарегистрировать плагин. Плагин должен иметь атрибут name.

Parameters:

Name	Type	Description	Default
plugin	Any	Объект или класс регистрируемого плагина.	required

SecretProviderPlugin

Bases: BasePlugin, SecretProvider

Интерфейс для плагина-провайдера секретов. Позволяет подключить стороннее хранилище секретов (например, AWS Secrets Manager, Vault).

register_plugin(plugin)

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

Parameters:

Name	Type	Description	Default
plugin	Any	Объект или класс регистрируемого плагина.	required

options: members:

• register_plugin
• registry
• BasePlugin
• SecretProviderPlugin
• ConfigProviderPlugin
• LoggerHandlerPlugin
• MetricsPlugin

Модуль http (HTTP-клиент и отказоустойчивость)

chutils.http

chutils.http — Лёгковесный HTTP-клиент с батареями.

Предоставляет синхронный и асинхронный HTTP-клиенты с встроенной поддержкой отказоустойчивости, трассировки и маскирования секретов.

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

from chutils.http import HttpClient, ResiliencePolicy

policy = ResiliencePolicy(retries=3, timeout=10.0)
with HttpClient(base_url="https://api.example.com", policy=policy) as client:
    resp = client.get("/users/1")
    resp.raise_for_status()
    data = resp.json()

Standalone-функции:

from chutils import http
resp = http.get("https://httpbin.org/get")

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

from chutils.http import AsyncHttpClient

async with AsyncHttpClient(base_url="https://api.example.com") as client:
    resp = await client.get("/status")

AsyncEventStreamClient

Асинхронный клиент для HTTP Streaming и SSE.

AsyncHttpClient

Асинхронный HTTP-клиент на базе httpx.AsyncClient.

Требует установленного httpx. При его отсутствии вызывает OptionalDependencyError при инициализации.

Parameters:

Name	Type	Description	Default
base_url	str	Базовый URL-префикс для всех запросов.	''
default_headers	dict[str, str] | None	Заголовки по умолчанию.	None
timeout	float | None	Таймаут запросов в секундах.	30.0
policy	ResiliencePolicy | None	Политика отказоустойчивости.	None
sensitive_headers	set[str] | None	Дополнительные заголовки для маскирования.	None

Example

from chutils.http import AsyncHttpClient, ResiliencePolicy

policy = ResiliencePolicy(retries=2, timeout=5.0)
async with AsyncHttpClient(base_url="https://api.example.com", policy=policy) as client:
    resp = await client.get("/status")
    resp.raise_for_status()

__aenter__() async

Поддержка async-контекстного менеджера.

Returns:

Type	Description
AsyncHttpClient	Сам экземпляр клиента.

__aexit__(*args) async

Закрывает async-клиент при выходе из контекстного менеджера.

__init__(*, base_url='', default_headers=None, timeout=30.0, policy=None, sensitive_headers=None)

Инициализирует AsyncHttpClient.

Parameters:

Name	Type	Description	Default
base_url	str	Базовый URL для всех запросов.	''
default_headers	dict[str, str] | None	Заголовки по умолчанию.	None
timeout	float | None	Таймаут в секундах.	30.0
policy	ResiliencePolicy | None	Политика отказоустойчивости.	None
sensitive_headers	set[str] | None	Имена заголовков для маскирования.	None

Raises:

Type	Description
OptionalDependencyError	Если httpx не установлен.

aclose() async

Закрывает async-клиент и освобождает ресурсы.

delete(path, *, headers=None, timeout=None) async

Выполняет async DELETE-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

get(path, *, headers=None, timeout=None) async

Выполняет async GET-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

patch(path, *, headers=None, json_data=None, data=None, timeout=None) async

Выполняет async PATCH-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела.	None
data	bytes | str | None	Сырое тело.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

post(path, *, headers=None, json_data=None, data=None, timeout=None) async

Выполняет async POST-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела.	None
data	bytes | str | None	Сырое тело.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

put(path, *, headers=None, json_data=None, data=None, timeout=None) async

Выполняет async PUT-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела.	None
data	bytes | str | None	Сырое тело.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

request(method, path, *, headers=None, json_data=None, data=None, timeout=None) async

Выполняет асинхронный HTTP-запрос.

Parameters:

Name	Type	Description	Default
method	str	HTTP-метод (GET, POST, PUT, DELETE, PATCH).	required
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела.	None
data	bytes | str | None	Сырое тело запроса.	None
timeout	float | None	Таймаут для этого конкретного запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

AsyncWebSocketClient

Асинхронный клиент для WebSockets.

connect() async

Устанавливает асинхронное соединение по WebSocket.

recv() async

Принимает текстовое или бинарное сообщение из WebSocket.

Returns:

Type	Description
str | bytes	Принятое сообщение.

send(message) async

Отправляет текстовое или бинарное сообщение через WebSocket.

Parameters:

Name	Type	Description	Default
message	str | bytes	Сообщение для отправки.	required

EventStreamClient

Синхронный клиент-обертка для HTTP Streaming и SSE.

HttpClient

Синхронный HTTP-клиент с батареями (httpx + fallback на urllib).

При наличии httpx использует его как транспорт. При отсутствии — прозрачно переключается на встроенный urllib.request, выводя единоразовое предупреждение в лог.

Parameters:

Name	Type	Description	Default
base_url	str	Базовый URL-префикс для всех запросов.	''
default_headers	dict[str, str] | None	Заголовки, добавляемые к каждому запросу.	None
timeout	float | None	Таймаут запросов по умолчанию в секундах.	30.0
policy	ResiliencePolicy | None	Политика отказоустойчивости (retry, semaphore и т.д.).	None
sensitive_headers	set[str] | None	Дополнительные заголовки для маскирования в логах.	None

Example

from chutils.http import HttpClient, ResiliencePolicy

policy = ResiliencePolicy(retries=3, timeout=10.0)
with HttpClient(base_url="https://api.example.com", policy=policy) as client:
    resp = client.get("/users/1")
    resp.raise_for_status()
    data = resp.json()

__enter__()

Поддержка контекстного менеджера.

Returns:

Type	Description
HttpClient	Сам экземпляр клиента.

__exit__(*args)

Закрывает клиент при выходе из контекстного менеджера.

__init__(*, base_url='', default_headers=None, timeout=30.0, policy=None, sensitive_headers=None)

Инициализирует HttpClient.

Parameters:

Name	Type	Description	Default
base_url	str	Базовый URL для всех запросов.	''
default_headers	dict[str, str] | None	Заголовки по умолчанию для каждого запроса.	None
timeout	float | None	Таймаут в секундах.	30.0
policy	ResiliencePolicy | None	Политика отказоустойчивости.	None
sensitive_headers	set[str] | None	Имена заголовков для маскирования в логах.	None

close()

Закрывает клиент и освобождает ресурсы.

delete(path, *, headers=None, timeout=None)

Выполняет DELETE-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

get(path, *, headers=None, timeout=None)

Выполняет GET-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

patch(path, *, headers=None, json_data=None, data=None, timeout=None)

Выполняет PATCH-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела.	None
data	bytes | str | None	Сырое тело.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

post(path, *, headers=None, json_data=None, data=None, timeout=None)

Выполняет POST-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела.	None
data	bytes | str | None	Сырое тело.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

put(path, *, headers=None, json_data=None, data=None, timeout=None)

Выполняет PUT-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела.	None
data	bytes | str | None	Сырое тело.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

request(method, path, *, headers=None, json_data=None, data=None, timeout=None)

Выполняет HTTP-запрос.

Parameters:

Name	Type	Description	Default
method	str	HTTP-метод (GET, POST, PUT, DELETE, PATCH).	required
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела запроса.	None
data	bytes | str | None	Сырое тело запроса (bytes или str).	None
timeout	float | None	Таймаут для этого конкретного запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

HttpResponse dataclass

Ответ HTTP-запроса.

Attributes:

Name	Type	Description
status_code	int	HTTP-статус-код ответа.
headers	dict[str, str]	Заголовки ответа.
content	bytes	Тело ответа в байтах.
elapsed	float	Время выполнения запроса в секундах.
url	str	Итоговый URL (с учётом редиректов).

text property

Тело ответа в виде строки UTF-8.

Returns:

Type	Description
str	Декодированное тело ответа.

json()

Десериализует тело ответа как JSON.

Returns:

Type	Description
object	Распарсенный JSON-объект.

Raises:

Type	Description
ValueError	Если тело не является корректным JSON.

raise_for_status()

Вызывает исключение, если статус-код указывает на ошибку (4xx / 5xx).

Raises:

Type	Description
HttpClientError	Если статус-код >= 400.

ResiliencePolicy

Политика отказоустойчивости: retry, timeout, semaphore, circuit breaker.

Применяется к произвольным синхронным (apply_sync) и асинхронным (apply_async) вызовам для обеспечения надёжности.

Attributes:

Name	Type	Description
retries		Количество повторных попыток после первого отказа.
retry_delay		Базовая задержка (сек.) между попытками.
retry_backoff		Множитель задержки для экспоненциального отступа.
retry_jitter		Добавлять ли случайный шум к задержке.
retry_exceptions		Кортеж классов исключений, при которых выполняется повтор.
retry_on_status_codes		Набор HTTP-статус-кодов, при которых выполняется повтор.
timeout		Максимальное время выполнения вызова (сек.); None — без ограничения.
max_concurrency		Максимальное число одновременных вызовов; None — без ограничения.
cb_failure_threshold		Порог отказов для размыкания Circuit Breaker.
cb_recovery_timeout		Время (сек.) до попытки восстановления Circuit Breaker.

Example

policy = ResiliencePolicy(retries=3, timeout=5.0, max_concurrency=10)
result = policy.apply_sync(requests.get, url)

__init__(*, retries=3, retry_delay=0.5, retry_backoff=2.0, retry_jitter=False, retry_exceptions=(Exception,), retry_on_status_codes=(429, 500, 502, 503, 504), timeout=None, max_concurrency=None, cb_failure_threshold=5, cb_recovery_timeout=30.0)

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

Parameters:

Name	Type	Description	Default
retries	int	Количество повторных попыток после первого отказа.	3
retry_delay	float	Базовая задержка между попытками в секундах.	0.5
retry_backoff	float	Множитель задержки для экспоненциального отступа.	2.0
retry_jitter	bool	Если True, добавляет случайный шум к задержке.	False
retry_exceptions	tuple[type[Exception], ...]	Кортеж классов исключений, при которых выполняется retry.	(Exception,)
retry_on_status_codes	tuple[int, ...]	HTTP-статус-коды для retry (используется с extractor).	(429, 500, 502, 503, 504)
timeout	float | None	Максимальное время выполнения вызова в секундах.	None
max_concurrency	int | None	Максимальное число одновременных вызовов.	None
cb_failure_threshold	int	Порог отказов для открытия Circuit Breaker.	5
cb_recovery_timeout	float	Пауза перед попыткой восстановления Circuit Breaker.	30.0

apply_async(func, *args, http_error_extractor=None, **kwargs) async

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

Оборачивает await func(*args, **kwargs) в retry, timeout и semaphore.

Parameters:

Name	Type	Description	Default
func	Callable[..., object]	Асинхронный вызываемый объект (coroutine function).	required
*args	object	Позиционные аргументы для func.	()
http_error_extractor	Callable[[Exception], int] | None	Опциональная функция для извлечения HTTP-статус-кода из пойманного исключения.	None
**kwargs	object	Именованные аргументы для func.	{}

Returns:

Type	Description
object	Результат await func(*args, **kwargs).

Raises:

Type	Description
ChutilsTimeoutError	При превышении timeout.
CircuitBreakerOpenError	Если Circuit Breaker разомкнут.
Exception	Последнее перехваченное исключение после исчерпания retry.

apply_sync(func, *args, http_error_extractor=None, **kwargs)

Применяет политику к синхронному вызову.

Оборачивает func(*args, **kwargs) в retry, timeout и semaphore.

Parameters:

Name	Type	Description	Default
func	Callable[..., object]	Вызываемый объект.	required
*args	object	Позиционные аргументы для func.	()
http_error_extractor	Callable[[Exception], int] | None	Опциональная функция для извлечения HTTP-статус-кода из пойманного исключения. Используется для retry по retry_on_status_codes.	None
**kwargs	object	Именованные аргументы для func.	{}

Returns:

Type	Description
object	Результат func(*args, **kwargs).

Raises:

Type	Description
ChutilsTimeoutError	При превышении timeout.
CircuitBreakerOpenError	Если Circuit Breaker разомкнут.
Exception	Последнее перехваченное исключение после исчерпания retry.

ServerSentEvent dataclass

Представляет собой отдельное событие Server-Sent Events (SSE).

UrllibFallbackClient

Синхронный HTTP-клиент на базе urllib.request.

Используется как fallback, когда httpx не доступен. Поддерживает интеграцию с ResiliencePolicy для retry, timeout и semaphore.

Parameters:

Name	Type	Description	Default
base_url	str	Базовый URL, который будет префиксом для всех запросов.	''
default_headers	dict[str, str] | None	Заголовки по умолчанию для всех запросов.	None
timeout	float | None	Таймаут подключения и чтения в секундах.	30.0
policy	ResiliencePolicy | None	Политика отказоустойчивости.	None
sensitive_headers	set[str] | None	Дополнительные заголовки для маскирования в логах.	None

Example

client = UrllibFallbackClient(
    base_url="https://api.example.com",
    default_headers={"Authorization": "Bearer token"},
    timeout=10.0,
)
response = client.get("/users/1")
response.raise_for_status()
data = response.json()

__enter__()

Поддержка контекстного менеджера.

Returns:

Type	Description
UrllibFallbackClient	Сам экземпляр клиента.

__exit__(*args)

Закрывает клиент при выходе из контекстного менеджера.

__init__(*, base_url='', default_headers=None, timeout=30.0, policy=None, sensitive_headers=None)

Инициализирует fallback HTTP-клиент.

Parameters:

Name	Type	Description	Default
base_url	str	Базовый URL для всех запросов.	''
default_headers	dict[str, str] | None	Заголовки, добавляемые к каждому запросу.	None
timeout	float | None	Таймаут в секундах (подключение + чтение).	30.0
policy	ResiliencePolicy | None	Политика отказоустойчивости (retry, semaphore и т.д.).	None
sensitive_headers	set[str] | None	Дополнительные имена заголовков для маскирования.	None

close()

Закрывает клиент (no-op для urllib-клиента, для совместимости API).

delete(path, *, headers=None, timeout=None)

Выполняет DELETE-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

get(path, *, headers=None, timeout=None)

Выполняет GET-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

patch(path, *, headers=None, json_data=None, data=None, timeout=None)

Выполняет PATCH-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела запроса.	None
data	bytes | str | None	Сырое тело запроса.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

post(path, *, headers=None, json_data=None, data=None, timeout=None)

Выполняет POST-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела запроса.	None
data	bytes | str | None	Сырое тело запроса.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

put(path, *, headers=None, json_data=None, data=None, timeout=None)

Выполняет PUT-запрос.

Parameters:

Name	Type	Description	Default
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки.	None
json_data	object | None	Данные для JSON-тела запроса.	None
data	bytes | str | None	Сырое тело запроса.	None
timeout	float | None	Таймаут запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

request(method, path, *, headers=None, json_data=None, data=None, timeout=None)

Выполняет HTTP-запрос с заданным методом.

Parameters:

Name	Type	Description	Default
method	str	HTTP-метод (GET, POST, PUT, DELETE, PATCH).	required
path	str	Путь или абсолютный URL.	required
headers	dict[str, str] | None	Дополнительные заголовки запроса.	None
json_data	object | None	Данные для сериализации в JSON-тело запроса.	None
data	bytes | str | None	Сырое тело запроса (bytes или str).	None
timeout	float | None	Таймаут для этого конкретного запроса.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

Raises:

Type	Description
HttpClientError	При сетевой ошибке.
ValueError	Если переданы одновременно json_data и data.

WebSocketClient

Синхронный клиент-обертка для WebSockets.

connect()

Устанавливает синхронное соединение по WebSocket.

recv()

Принимает текстовое или бинарное сообщение из WebSocket.

Returns:

Type	Description
str | bytes	Принятое сообщение.

send(message)

Отправляет текстовое или бинарное сообщение через WebSocket.

Parameters:

Name	Type	Description	Default
message	str | bytes	Сообщение для отправки.	required

create_http_span(method, url, tracer_name='chutils.http')

Создаёт OTEL-спан для исходящего HTTP-запроса.

Если OpenTelemetry не установлен, возвращает None.

Parameters:

Name	Type	Description	Default
method	str	HTTP-метод (GET, POST и т.д.).	required
url	str	Полный URL запроса.	required
tracer_name	str	Имя трейсера (используется для группировки спанов).	'chutils.http'

Returns:

Type	Description
object | None	Контекстный менеджер спана или None, если OTEL недоступен.

Example

from chutils.http.tracing import create_http_span

span = create_http_span("GET", "https://api.example.com/users")
if span is not None:
    with span:
        response = requests.get(url)

delete(url, *, headers=None, timeout=None, policy=None)

Выполняет DELETE-запрос.

Parameters:

Name	Type	Description	Default
url	str	Абсолютный URL запроса.	required
headers	dict[str, str] | None	Дополнительные HTTP-заголовки.	None
timeout	float | None	Таймаут запроса в секундах.	None
policy	ResiliencePolicy | None	Политика отказоустойчивости.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

get(url, *, headers=None, timeout=None, policy=None)

Выполняет GET-запрос.

Создаёт временный HttpClient, выполняет запрос и возвращает ответ.

Parameters:

Name	Type	Description	Default
url	str	Абсолютный URL запроса.	required
headers	dict[str, str] | None	Дополнительные HTTP-заголовки.	None
timeout	float | None	Таймаут запроса в секундах.	None
policy	ResiliencePolicy | None	Политика отказоустойчивости (retry, timeout, semaphore).	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse с телом, заголовками и статус-кодом ответа.

Example

from chutils import http

resp = http.get("https://httpbin.org/get", timeout=5.0)
resp.raise_for_status()
data = resp.json()

inject_trace_headers(headers)

Инжектирует W3C Trace Context заголовки в словарь заголовков запроса.

Если OpenTelemetry не установлен или трассировка не настроена, возвращает заголовки без изменений.

Инжектируемые заголовки: - traceparent: идентификаторы trace и span (формат W3C). - tracestate: дополнительное состояние вендора (опционально).

Parameters:

Name	Type	Description	Default
headers	dict[str, str]	Исходный словарь HTTP-заголовков запроса.	required

Returns:

Type	Description
dict[str, str]	Обновлённый словарь с добавленными заголовками трассировки
dict[str, str]	(или оригинальный, если OTEL недоступен).

Example

from chutils.http.tracing import inject_trace_headers

headers = {"Authorization": "Bearer token"}
headers = inject_trace_headers(headers)
# headers теперь содержит "traceparent" если OTEL активен

patch(url, *, headers=None, json_data=None, data=None, timeout=None, policy=None)

Выполняет PATCH-запрос.

Parameters:

Name	Type	Description	Default
url	str	Абсолютный URL запроса.	required
headers	dict[str, str] | None	Дополнительные HTTP-заголовки.	None
json_data	object | None	Данные для JSON-тела запроса.	None
data	bytes | str | None	Сырое тело запроса.	None
timeout	float | None	Таймаут запроса в секундах.	None
policy	ResiliencePolicy | None	Политика отказоустойчивости.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

post(url, *, headers=None, json_data=None, data=None, timeout=None, policy=None)

Выполняет POST-запрос.

Parameters:

Name	Type	Description	Default
url	str	Абсолютный URL запроса.	required
headers	dict[str, str] | None	Дополнительные HTTP-заголовки.	None
json_data	object | None	Данные для сериализации в JSON-тело запроса.	None
data	bytes | str | None	Сырое тело запроса (bytes или str).	None
timeout	float | None	Таймаут запроса в секундах.	None
policy	ResiliencePolicy | None	Политика отказоустойчивости.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

Example

resp = http.post(
    "https://api.example.com/users",
    json_data={"name": "Alice", "email": "alice@example.com"},
)
resp.raise_for_status()

put(url, *, headers=None, json_data=None, data=None, timeout=None, policy=None)

Выполняет PUT-запрос.

Parameters:

Name	Type	Description	Default
url	str	Абсолютный URL запроса.	required
headers	dict[str, str] | None	Дополнительные HTTP-заголовки.	None
json_data	object | None	Данные для JSON-тела запроса.	None
data	bytes | str | None	Сырое тело запроса.	None
timeout	float | None	Таймаут запроса в секундах.	None
policy	ResiliencePolicy | None	Политика отказоустойчивости.	None

Returns:

Type	Description
HttpResponse	Объект HttpResponse.

options: members:

• HttpClient
• AsyncHttpClient
• ResiliencePolicy
• HttpResponse
• get
• post
• put
• delete
• patch