Обновлены пути к библиотекам в Dockerfile для соответствия новой версии Python. Исправлены все тесты, теперь все проходят
4.8 KiB
4.8 KiB
description, globs
| description | globs | ||
|---|---|---|---|
| Работа с зависимостями, утилитами, метриками и внешними сервисами |
|
Зависимости и утилиты
BaseDependencyFactory
Центральный класс для управления зависимостями проекта.
Использование
from helper_bot.utils.base_dependency_factory import get_global_instance
# Получение глобального экземпляра
bdf = get_global_instance()
# Доступ к зависимостям
db = bdf.get_db() # AsyncBotDB
settings = bdf.get_settings() # dict с настройками
s3_storage = bdf.get_s3_storage() # S3StorageService или None
Структура settings
Настройки загружаются из .env и структурированы:
settings = {
'Telegram': {
'bot_token': str,
'listen_bot_token': str,
'preview_link': bool,
'main_public': str,
'group_for_posts': int,
'important_logs': int,
...
},
'Settings': {
'logs': bool,
'test': bool
},
'Metrics': {
'host': str,
'port': int
},
'S3': {
'enabled': bool,
'endpoint_url': str,
'access_key': str,
'secret_key': str,
'bucket_name': str,
'region': str
}
}
Метрики
Декораторы метрик
Используйте декораторы из helper_bot.utils.metrics:
from helper_bot.utils.metrics import track_time, track_errors, db_query_time
@track_time("method_name", "module_name")
@track_errors("module_name", "method_name")
async def some_method():
"""Метод с отслеживанием времени и ошибок."""
...
@db_query_time("method_name", "table_name", "operation")
async def db_method():
"""Метод БД с отслеживанием времени запросов."""
...
Доступ к метрикам
from helper_bot.utils.metrics import metrics
# Метрики доступны через Prometheus на порту из settings['Metrics']['port']
Rate Limiting
RateLimiter
Используется для ограничения частоты запросов:
from helper_bot.utils.rate_limiter import RateLimiter
limiter = RateLimiter(...)
if await limiter.is_allowed(user_id):
# Разрешить действие
...
else:
# Отклонить действие
...
RateLimitMiddleware
Автоматически применяет rate limiting ко всем запросам через middleware.
S3 Storage
S3StorageService
Используется для хранения медиафайлов:
from helper_bot.utils.s3_storage import S3StorageService
# Получение через BaseDependencyFactory
s3_storage = bdf.get_s3_storage()
if s3_storage:
# Загрузка файла
url = await s3_storage.upload_file(file_path, object_key)
# Удаление файла
await s3_storage.delete_file(object_key)
Проверка доступности
Всегда проверяйте, что S3 включен:
s3_storage = bdf.get_s3_storage()
if s3_storage:
# Работа с S3
...
else:
# Fallback логика
...
Утилиты
helper_func.py
Содержит вспомогательные функции для работы с:
- Датами и временем
- Форматированием данных
- Валидацией
- Преобразованием данных
Используйте эти функции вместо дублирования логики.
Конфигурация
rate_limit_config.py
Конфигурация rate limiting находится в helper_bot/config/rate_limit_config.py.
Используйте конфигурацию вместо хардкода значений.
Best Practices
- Всегда получайте зависимости через BaseDependencyFactory - не создавайте экземпляры напрямую
- Используйте декораторы метрик для всех важных методов
- Проверяйте доступность внешних сервисов (S3) перед использованием
- Используйте утилиты из
helper_func.pyвместо дублирования кода - Читайте настройки из settings вместо хардкода значений
- Логируйте важные операции с внешними сервисами