PassiveIncome.CommonPackage

Набор общих инструментов для Python-приложений: настройки на основе Pydantic, безопасный вывод конфигурации, настройка стандартного модуля logging и клиент RabbitMQ.

Возможности

• AppSettings — базовый класс для настроек приложения.
• safe_dump() — представление настроек с маскированием чувствительных значений.
• key_masking() — рекурсивное маскирование словарей, списков и кортежей.
• LoggingSettings и configure_logging() — глобальная настройка уровня и обработчиков логирования.
• RabbitProducer и RabbitConsumer — публикация и обработка типизированных сообщений RabbitMQ.

Требования

• Python 3.13 или новее.

Установка

pip install passiveincome-commonpackage==1.1.0

Для разработки из клона репозитория установите пакет и dev-зависимости:

python -m pip install -e ".[dev]"

Настройки приложения

Наследуйте AppSettings и объявите поля, нужные вашему приложению. Как и другие модели Pydantic Settings, значения могут поступать из переменных окружения.

from common_package import AppSettings


class ServiceSettings(AppSettings):
	service_name: str = "worker"
	database_password: str
	api_token: str


settings = ServiceSettings()
print(settings.service_name)

Для обязательных полей задайте соответствующие переменные окружения, например DATABASE_PASSWORD и API_TOKEN.

Безопасный вывод

Метод safe_dump() возвращает словарь настроек, в котором значения полей с чувствительными словами в имени заменены на ***. Сопоставление нечувствительно к регистру и учитывает вхождение слова как подстроки.

print(settings.safe_dump())
# {'service_name': 'worker', 'database_password': '***', 'api_token': '***'}

По умолчанию маскируются имена, содержащие password, passwd, pwd, secret, token, key, credential, auth, session, cookie, salt или hash.

Для произвольных структур можно использовать key_masking(data, key_words, mask="***"). Функция обходит вложенные словари, списки и кортежи, не изменяя исходные данные.

Логирование

Создайте LoggingSettings и передайте их в configure_logging(). Настройка применяется к root logger, поэтому влияет на логгеры приложения, использующие стандартную иерархию Python logging. Текущие обработчики root logger заменяются.

import logging

from common_package import LoggingSettings, configure_logging


settings = LoggingSettings()
configure_logging(settings)

logger = logging.getLogger(__name__)
logger.info("Приложение запущено")

LoggingSettings загружает значения окружения с префиксом LOGGING_. Например:

export LOGGING_LEVEL=DEBUG
export LOGGING_STREAM=true
export LOGGING_FILE_PATH=logs/application.log
export LOGGING_MAX_BYTES=10485760
export LOGGING_BACKUP_COUNT=5

Если одновременно отключить stream и не задать file_path, устанавливается NullHandler.

RabbitMQ

RabbitProducer объявляет exchange, очередь и binding между ними. RabbitConsumer объявляет очередь и передаёт десериализованные сообщения асинхронному обработчику. Укажите один и тот же queue_name в настройках producer и consumer. Сообщение должно быть экземпляром вашей модели, реализующей BaseRabbitMessage; consumer использует её from_bytes() для десериализации.

Пример настройки consumer и запуска обработки:

import asyncio

from common_package.rabbit.consumer import RabbitConsumer
from common_package.rabbit.models import BaseRabbitMessage
from common_package.rabbit.settings import RabbitConsumerSettings


async def handle(message: BaseRabbitMessage) -> None:
	print(message.to_dict())


async def main() -> None:
	settings = RabbitConsumerSettings(
		host="localhost",
		port=5672,
		user="guest",
		password="guest",
		queue_name="events",
	)
	consumer = RabbitConsumer(settings, message_type=YourMessage)
	await consumer.setup()
	await consumer.consume(handle)


asyncio.run(main())

Замените YourMessage на свою реализацию BaseRabbitMessage. Consumer подтверждает сообщение после успешного завершения обработчика. Если десериализация или обработчик выбрасывает исключение, сообщение возвращается в очередь для повторной доставки, поэтому для постоянно ошибочных сообщений рекомендуется настроить ограничение повторов или dead-letter очередь.

Для публикации создайте RabbitProducer с RabbitProducerSettings, вызовите await producer.setup(), затем await producer.publish(your_message). Настройки producer включают подключение, ExchangeSettings, QueueSettings и routing_key; очередь должна соответствовать очереди consumer.

Запуск тестов

После установки dev-зависимостей запустите из корня репозитория:

python -m pytest