Архитектура телеграм-ботов на Python и Феномен 15 миллиардов сообщений (2)

import hashlib
from typing import List, Optional
 
class ConsistentHashRing:
    def __init__(self, nodes: List[str], virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self.ring = {}
        self.sorted_keys = []
        
        for node in nodes:
            self.add_node(node)
    
    def _hash(self, key: str) -> int:
        # Хешируем в 32-битное число
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def add_node(self, node: str):
        # Добавляем виртуальные ноды для равномерности
        for i in range(self.virtual_nodes):
            virtual_key = f"{node}:{i}"
            hash_val = self._hash(virtual_key)
            self.ring[hash_val] = node
        
        # Пересортировываем ключи
        self.sorted_keys = sorted(self.ring.keys())
    
    def get_node(self, user_id: int) -> str:
        # Находим шард для конкретного юзера
        if not self.ring:
            return None
        
        key_hash = self._hash(str(user_id))
        
        # Ищем первый шард по часовой стрелке
        for ring_key in self.sorted_keys:
            if key_hash <= ring_key:
                return self.ring[ring_key]
        
        # Если прошли весь круг - возвращаем первый
        return self.ring[self.sorted_keys[0]]
 
# Использование
hash_ring = ConsistentHashRing([
    'db-shard-01.example.com',
    'db-shard-02.example.com',
    'db-shard-03.example.com'
])
 
# Каждый юзер попадает на предсказуемый шард
shard = hash_ring.get_node(user_id=12345)
# shard = 'db-shard-02.example.com'

class ShardedDatabase:
    def __init__(self, shard_configs: dict):
        self.shards = {}
        self.rebalancing = {}  # Временные маппинги при миграции
        
        for shard_id, config in shard_configs.items():
            self.shards[shard_id] = asyncpg.create_pool(config['dsn'])
    
    def get_shard_id(self, user_id: int) -> str:
        # Проверяем идёт ли миграция для этого юзера
        if user_id in self.rebalancing:
            return self.rebalancing[user_id]
        
        # Стандартное распределение
        shard_count = len(self.shards)
        return f"shard_{user_id % shard_count}"
    
    async def get_user(self, user_id: int):
        shard_id = self.get_shard_id(user_id)
        pool = self.shards[shard_id]
        
        async with pool.acquire() as conn:
            # Пытаемся найти на целевом шарде
            user = await conn.fetchrow(
                "SELECT * FROM users WHERE id = $1",
                user_id
            )
            
            if not user and user_id in self.rebalancing:
                # Миграция в процессе - пробуем старый шард
                old_shard = self.rebalancing[user_id]
                async with self.shards[old_shard].acquire() as old_conn:
                    user = await old_conn.fetchrow(
                        "SELECT * FROM users WHERE id = $1",
                        user_id
                    )
            
            return user
    
    async def migrate_user(self, user_id: int, target_shard: str):
        source_shard = self.get_shard_id(user_id)
        
        if source_shard == target_shard:
            return  # Уже на нужном шарде
        
        # Помечаем что идёт миграция
        self.rebalancing[user_id] = source_shard
        
        try:
            # Копируем данные на новый шард
            async with self.shards[source_shard].acquire() as src_conn:
                user_data = await src_conn.fetchrow(
                    "SELECT * FROM users WHERE id = $1",
                    user_id
                )
                
                if user_data:
                    async with self.shards[target_shard].acquire() as dst_conn:
                        # Вставляем на новый шард
                        await dst_conn.execute(
                            """INSERT INTO users (id, username, data) 
                               VALUES ($1, $2, $3)""",
                            user_data['id'],
                            user_data['username'],
                            user_data['data']
                        )
                    
                    # Удаляем со старого
                    await src_conn.execute(
                        "DELETE FROM users WHERE id = $1",
                        user_id
                    )
            
            # Обновляем маппинг
            del self.rebalancing[user_id]
        except Exception as e:
            # При ошибке откатываем маркер миграции
            logger.error(f"Migration failed for user {user_id}: {e}")
            del self.rebalancing[user_id]
            raise

frontend bot_frontend
    bind *:443 ssl crt /etc/ssl/bot.pem
    default_backend bot_backends
 
backend bot_backends
    balance source  # Sticky sessions по IP
    hash-type consistent
    
    option httpchk GET /health
    http-check expect status 200
    
    server bot1 10.0.1.10:8443 check inter 2000 rise 2 fall 3
    server bot2 10.0.1.11:8443 check inter 2000 rise 2 fall 3
    server bot3 10.0.1.12:8443 check inter 2000 rise 2 fall 3

# Command model - нормализованная структура
class CreateOrderCommand:
    async def execute(self, user_id: int, items: list):
        async with db.transaction():
            # Проверяем остатки
            for item in items:
                stock = await db.fetchval(
                    "SELECT quantity FROM inventory WHERE product_id = $1 FOR UPDATE",
                    item['product_id']
                )
                if stock < item['quantity']:
                    raise InsufficientStock()
            
            # Создаём заказ
            order_id = await db.fetchval(
                "INSERT INTO orders (user_id, status, created_at) VALUES ($1, $2, NOW()) RETURNING id",
                user_id, 'pending'
            )
            
            # Записываем позиции
            for item in items:
                await db.execute(
                    "INSERT INTO order_items (order_id, product_id, quantity, price) VALUES ($1, $2, $3, $4)",
                    order_id, item['product_id'], item['quantity'], item['price']
                )
                
                # Обновляем остатки
                await db.execute(
                    "UPDATE inventory SET quantity = quantity - $1 WHERE product_id = $2",
                    item['quantity'], item['product_id']
                )
            
            # Публикуем событие для обновления read model
            await event_bus.publish('order.created', {
                'order_id': order_id,
                'user_id': user_id,
                'items': items
            })
            
            return order_id
 
# Query model - денормализованная выборка
class OrderQueryService:
    async def get_user_orders(self, user_id: int, limit: int = 10):
        # Читаем из материализованного view или отдельной таблицы
        orders = await read_db.fetch("""
            SELECT 
                o.id,
                o.status,
                o.created_at,
                o.total_amount,
                json_agg(json_build_object(
                    'product_name', p.name,
                    'quantity', oi.quantity,
                    'price', oi.price
                )) as items
            FROM orders_read_model o
            JOIN order_items_read_model oi ON o.id = oi.order_id
            JOIN products_cache p ON oi.product_id = p.id
            WHERE o.user_id = $1
            GROUP BY o.id
            ORDER BY o.created_at DESC
            LIMIT $2
        """, user_id, limit)
        
        return orders

# Катастрофа для async кода
async def bad_handler(message: Message):
    # Блокирует весь event loop на секунды
    response = requests.get("https://api.example.com/data")
    result = heavy_cpu_function(response.json())
    
    await message.answer(str(result))
 
# Правильный подход
async def good_handler(message: Message):
    # Async HTTP-клиент не блокирует
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/data") as resp:
            data = await resp.json()
    
    # CPU-bound операция в executor
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, heavy_cpu_function, data)
    
    await message.answer(str(result))

import cProfile
import pstats
from pstats import SortKey
 
def profile_bot_handler():
    profiler = cProfile.Profile()
    profiler.enable()
    
    # Здесь выполняется код бота
    # Например, обработка 1000 сообщений
    
    profiler.disable()
    
    # Сохраняем результаты
    profiler.dump_stats('bot_profile.prof')
    
    # Анализируем топ функций по времени
    stats = pstats.Stats(profiler)
    stats.sort_stats(SortKey.CUMULATIVE)
    stats.print_stats(20)  # Топ 20 самых медленных

from line_profiler import LineProfiler
 
@profile  # Магический декоратор line_profiler
async def process_order(order_data: dict):
    # Валидация
    validated = OrderSchema(**order_data)
    
    # Проверка наличия
    available = await check_inventory(validated.items)
    
    # Создание заказа
    order_id = await db.create_order(validated)
    
    # Резервирование товаров
    await reserve_items(order_id, validated.items)
    
    # Отправка уведомления
    await notify_user(validated.user_id, order_id)
    
    return order_id
 
# Запуск с профилированием
# kernprof -l -v script.py

from memory_profiler import profile
 
@profile
def load_user_data(user_ids: list):
    # Загружаем данные всех юзеров
    users = []
    for uid in user_ids:
        user = fetch_user_from_db(uid)  # +50KB памяти
        user['cache'] = build_user_cache(user)  # +200KB
        users.append(user)
    
    return users
 
# Line #    Mem usage    Increment  Occurrences   Line Contents
# ===============================================================
#     2    125.5 MiB    125.5 MiB           1       users = []
#     3    125.5 MiB      0.0 MiB        1001       for uid in user_ids:
#     4    175.8 MiB     50.3 MiB        1000           user = fetch_user_from_db(uid)
#     5    375.8 MiB    200.0 MiB        1000           user['cache'] = build_user_cache(user)

import tracemalloc
 
# Начинаем отслеживание
tracemalloc.start()
 
# Делаем snapshot до операции
snapshot1 = tracemalloc.take_snapshot()
 
# Подозрительная операция
await process_thousand_messages()
 
# Snapshot после
snapshot2 = tracemalloc.take_snapshot()
 
# Сравниваем
top_stats = snapshot2.compare_to(snapshot1, 'lineno')
 
for stat in top_stats[:10]:
    print(f"{stat.size_diff / 1024 / 1024:.1f} MB")
    print(f"  {stat.traceback}")

# Профилируем процесс бота
py-spy top --pid 12345
 
# Или записываем flamegraph
py-spy record -o profile.svg --pid 12345 --duration 60

# Утечка памяти гарантирована
class User:
    def __init__(self, user_id):
        self.user_id = user_id
        self.session = None
 
class Session:
    def __init__(self, user):
        self.user = user
        user.session = self  # Циклическая ссылка
 
# Создаём и забываем
def handle_user():
    user = User(123)
    session = Session(user)
    # Выходим из функции - user и session должны удалиться
    # Но refcount у обоих = 1 из-за взаимных ссылок
    # Reference counting не может их почистить
 
# После тысячи вызовов - тысяча пар висит в памяти

import weakref
 
class CacheManager:
    def __init__(self):
        # WeakValueDictionary автоматически чистит умершие объекты
        self._cache = weakref.WeakValueDictionary()
    
    def cache_user(self, user_id, user_obj):
        self._cache[user_id] = user_obj
    
    def get_user(self, user_id):
        return self._cache.get(user_id)
 
# Объект удалится когда никто кроме кеша не держит на него ссылку
manager = CacheManager()
user = User(123)
manager.cache_user(123, user)
 
# user жив потому что переменная существует
assert manager.get_user(123) is not None
 
del user  # Удалили последнюю сильную ссылку
import gc; gc.collect()
 
# Теперь объект исчез из кеша автоматически
assert manager.get_user(123) is None

# Тяжёлый объект с __dict__
class User:
    def __init__(self, user_id, username):
        self.user_id = user_id
        self.username = username
        self.balance = 0
 
# Лёгкий объект со slots
class OptimizedUser:
    __slots__ = ('user_id', 'username', 'balance')
    
    def __init__(self, user_id, username):
        self.user_id = user_id
        self.username = username
        self.balance = 0
 
import sys
print(sys.getsizeof(User(1, "test")))  # ~400 байт
print(sys.getsizeof(OptimizedUser(1, "test")))  # ~200 байт

import objgraph
import gc
 
# Snapshot перед операцией  
gc.collect()
before = objgraph.typestats()
 
# Подозрительная операция
for i in range(1000):
    process_user_session()
 
# Snapshot после
gc.collect()
after = objgraph.typestats()
 
# Находим что утекло
objgraph.show_growth()
# dict         +5000
# list         +3000
[H2]User         +1000[/H2]
 
# Показываем кто держит User объекты живыми
objgraph.show_backrefs(objgraph.by_type('User')[0], max_depth=5)

import asyncpg
 
async def create_order_atomic(user_id: int, items: list, total: float):
    async with db_pool.acquire() as conn:
        async with conn.transaction():
            # Проверяем и резервируем товары
            for item in items:
                stock = await conn.fetchval(
                    "SELECT quantity FROM inventory WHERE product_id = $1 FOR UPDATE",
                    item['product_id']
                )
                
                if stock < item['quantity']:
                    raise InsufficientStock(f"Not enough {item['name']}")
                
                await conn.execute(
                    "UPDATE inventory SET quantity = quantity - $1 WHERE product_id = $2",
                    item['quantity'], item['product_id']
                )
            
            # Списываем деньги
            balance = await conn.fetchval(
                "SELECT balance FROM users WHERE id = $1 FOR UPDATE",
                user_id
            )
            
            if balance < total:
                raise InsufficientFunds()
            
            await conn.execute(
                "UPDATE users SET balance = balance - $1 WHERE id = $2",
                total, user_id
            )
            
            # Создаём заказ
            order_id = await conn.fetchval(
                "INSERT INTO orders (user_id, total, status) VALUES ($1, $2, 'pending') RETURNING id",
                user_id, total
            )
            
            # Если хоть что-то упало - всё откатывается автоматически
            return order_id

from motor.motor_asyncio import AsyncIOMotorClient
 
# MongoDB - гибкая схема документа
mongo_client = AsyncIOMotorClient('mongodb://localhost:27017')
db = mongo_client.blog
 
async def create_article_flexible(article_data: dict):
    # Любая структура данных без миграций схемы
    result = await db.articles.insert_one({
        'title': article_data['title'],
        'content': article_data['content'],
        'author_id': article_data['author_id'],
        'published_at': datetime.now(),
        # Опциональные поля просто включаются если есть
        **{k: v for k, v in article_data.items() 
           if k not in ['title', 'content', 'author_id']}
    })
    return result.inserted_id
 
async def get_article(article_id):
    # Получаем документ целиком со всеми полями
    article = await db.articles.find_one({'_id': article_id})
    
    # Проверяем наличие опциональных полей в рантайме
    if 'gallery' in article:
        # Обрабатываем галерею
        pass
    
    if 'poll' in article:
        # Обрабатываем опрос
        pass
    
    return article

# Синхронный код - блокирует event loop
import psycopg2
 
async def get_user_bad(user_id: int):
    conn = psycopg2.connect(DATABASE_URL)  # Блокирующее соединение
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))  # Блокирует на 20ms
    result = cursor.fetchone()
    conn.close()
    return result
    # Event loop заморожен всё это время
 
# Асинхронный код - не блокирует
import asyncpg
 
async def get_user_good(user_id: int):
    async with pool.acquire() as conn:  # Async получение соединения
        result = await conn.fetchrow(
            "SELECT * FROM users WHERE id = $1", 
            user_id
        )  # Event loop свободен пока ждём
        return result

import asyncpg
 
class DatabaseManager:
    def __init__(self, dsn: str, min_size: int = 10, max_size: int = 20):
        self.dsn = dsn
        self.min_size = min_size
        self.max_size = max_size
        self.pool = None
    
    async def connect(self):
        # Создаём пул с минимальным и максимальным размером
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=self.min_size,  # Всегда держим минимум соединений
            max_size=self.max_size,  # Не создаём больше максимума
            command_timeout=60,      # Таймаут на выполнение команды
            max_queries=50000,       # Пересоздаём соединение после N запросов
            max_inactive_connection_lifetime=300  # Закрываем простаивающие
        )
    
    async def fetch_user(self, user_id: int):
        # Автоматически берёт соединение из пула
        async with self.pool.acquire() as conn:
            return await conn.fetchrow(
                "SELECT * FROM users WHERE id = $1",
                user_id
            )
        # Соединение вернулось в пул автоматически
 
    async def close(self):
        await self.pool.close()

# N+1 антипаттерн - катастрофически неэффективно
async def get_orders_with_users_bad():
    async with pool.acquire() as conn:
        # Один запрос получения заказов
        orders = await conn.fetch("SELECT * FROM orders LIMIT 100")
        
        # Плюс N запросов для каждого юзера - итого 101 запрос
        result = []
        for order in orders:
            user = await conn.fetchrow(
                "SELECT username FROM users WHERE id = $1",
                order['user_id']
            )
            result.append({**order, 'username': user['username']})
        
        return result
 
# Правильно - один запрос с JOIN
async def get_orders_with_users_good():
    async with pool.acquire() as conn:
        # Один запрос получает всё сразу
        return await conn.fetch("""
            SELECT 
                o.*,
                u.username
            FROM orders o
            JOIN users u ON o.user_id = u.id
            LIMIT 100
        """)

# Создание индексов при инициализации базы
async def setup_indexes(conn):
    # Индекс для поиска по username
    await conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_users_username 
        ON users(username)
    """)
    
    # Составной индекс для фильтрации заказов
    await conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_orders_user_status 
        ON orders(user_id, status, created_at DESC)
    """)
    
    # Частичный индекс только для активных заказов
    await conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_orders_active 
        ON orders(user_id, created_at DESC)
        WHERE status IN ('pending', 'processing')
    """)

# Неэффективно - N отдельных INSERT'ов
async def save_events_slow(events: list):
    async with pool.acquire() as conn:
        for event in events:
            await conn.execute(
                "INSERT INTO events (user_id, action, data) VALUES ($1, $2, $3)",
                event['user_id'], event['action'], event['data']
            )
 
# Эффективно - один batch INSERT
async def save_events_fast(events: list):
    async with pool.acquire() as conn:
        # executemany оптимизирован для батчей
        await conn.executemany(
            "INSERT INTO events (user_id, action, data) VALUES ($1, $2, $3)",
            [(e['user_id'], e['action'], e['data']) for e in events]
        )

import asyncio
from datetime import datetime, timedelta
 
class TelegramRateLimiter:
    def __init__(self, rate_per_second: int = 25):
        self.rate = rate_per_second
        self.tokens = rate_per_second
        self.updated_at = datetime.now()
        self.lock = asyncio.Lock()
    
    async def acquire(self):
        async with self.lock:
            now = datetime.now()
            elapsed = (now - self.updated_at).total_seconds()
            
            # Добавляем токены за прошедшее время
            self.tokens = min(
                self.rate,
                self.tokens + elapsed * self.rate
            )
            self.updated_at = now
            
            # Если нет токенов - ждём
            if self.tokens < 1:
                sleep_time = (1 - self.tokens) / self.rate
                await asyncio.sleep(sleep_time)
                self.tokens = 0
            else:
                self.tokens -= 1
 
limiter = TelegramRateLimiter(rate_per_second=25)
 
async def send_with_rate_limit(bot, chat_id: int, text: str):
    await limiter.acquire()
    return await bot.send_message(chat_id, text)
 
# Массовая рассылка без перегрузки API
async def broadcast_message(bot, user_ids: list, text: str):
    tasks = [
        send_with_rate_limit(bot, uid, text) 
        for uid in user_ids
    ]
    # Ограничиваем concurrency для экономии памяти
    results = []
    for chunk in chunks(tasks, 100):
        batch_results = await asyncio.gather(*chunk, return_exceptions=True)
        results.extend(batch_results)
    
    return results

import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import Optional
 
class Priority(Enum):
HIGH = 1
NORMAL = 2
LOW = 3
 
@dataclass
class QueuedMessage:
priority: Priority
chat_id: int
text: str
timestamp: float
 
class PriorityQueue:
def __init__(self, rate_limiter):
    self.high = asyncio.Queue()
    self.normal = asyncio.Queue()
    self.low = asyncio.Queue()
    self.limiter = rate_limiter
 
async def put(self, msg: QueuedMessage):
    # Кладём в нужную очередь по приоритету
    if msg.priority == Priority.HIGH:
        await self.high.put(msg)
    elif msg.priority == Priority.NORMAL:
        await self.normal.put(msg)
    else:
        await self.low.put(msg)
 
async def get(self) -> Optional[QueuedMessage]:
    # Берём из высокоприоритетной если есть
    if not self.high.empty():
        return await self.high.get()
    
    # Проверяем не близки ли к лимиту
    if self.limiter.tokens < 5:
        # Близко к исчерпанию - только high priority
        return None
    
    if not self.normal.empty():
        return await self.normal.get()
    
    # Low только когда запас большой
    if self.limiter.tokens > 10 and not self.low.empty():
        return await self.low.get()
    
    return None
 
# Worker который обрабатывает очередь
async def queue_worker(bot, queue: PriorityQueue):
while True:
    msg = await queue.get()
    if msg:
        await limiter.acquire()
        try:
            await bot.send_message(msg.chat_id, msg.text)
        except Exception as e:
            # При ошибке возвращаем в очередь
            if "retry" in str(e).lower():
                await asyncio.sleep(5)
                await queue.put(msg)
    else:
        await asyncio.sleep(0.1)

import structlog
 
logger = structlog.get_logger()
 
async def process_order(user_id: int, order_id: int):
    logger.info(
        "order_processing_started",
        user_id=user_id,
        order_id=order_id,
        timestamp=time.time()
    )
    
    try:
        result = await create_order(order_id)
        logger.info(
            "order_created_successfully",
            user_id=user_id,
            order_id=order_id,
            amount=result.total
        )
    except PaymentError as e:
        logger.error(
            "payment_failed",
            user_id=user_id,
            order_id=order_id,
            error=str(e),
            exc_info=True
        )
        raise

import contextvars
import structlog
 
user_context = contextvars.ContextVar('user_id', default=None)
 
def add_user_context(logger, method_name, event_dict):
    # Добавляем user_id из контекста если есть
    uid = user_context.get()
    if uid:
        event_dict['user_id'] = uid
    return event_dict
 
structlog.configure(
    processors=[
        add_user_context,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)
 
logger = structlog.get_logger()
 
async def handle_message(message: Message):
    # Устанавливаем контекст для всех логов ниже
    user_context.set(message.from_user.id)
    
    # Все эти логи автоматически содержат user_id
    logger.info("message_received", text=message.text)
    await process_command(message.text)
    logger.info("message_processed")

import re
 
class SensitiveDataFilter:
    PATTERNS = [
        (r'\b\d{16}\b', '[B][/B]-[B][/B]-[B][/B]-[B][/B]'),  # Номера карт
        (r'token["\s:=]+([^\s,"]+)', r'token: [REDACTED]'),  # Токены
        (r'password["\s:=]+([^\s,"]+)', r'password: [REDACTED]'),
        (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]')
    ]
    
    def __call__(self, logger, method_name, event_dict):
        # Маскируем чувствительные данные в message
        if 'message' in event_dict:
            msg = str(event_dict['message'])
            for pattern, replacement in self.PATTERNS:
                msg = re.sub(pattern, replacement, msg)
            event_dict['message'] = msg
        return event_dict
 
structlog.configure(
    processors=[
        SensitiveDataFilter(),
        structlog.processors.JSONRenderer()
    ]
)

import os
import logging.config
 
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
 
LOGGING = {
    'version': 1,
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': LOG_LEVEL,
        },
        'file': {
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'filename': 'logs/bot.log',
            'when': 'midnight',
            'interval': 1,
            'backupCount': 7,
            'level': 'INFO',
        }
    },
    'root': {
        'level': LOG_LEVEL,
        'handlers': ['console', 'file']
    }
}
 
logging.config.dictConfig(LOGGING)

from prometheus_client import Counter, Gauge, Histogram, start_http_server
import time
 
# Счетчики событий
messages_total = Counter(
    'bot_messages_total',
    'Total messages processed',
    ['command', 'status']  # Лейблы для группировки
)
 
# Текущие значения
active_users = Gauge('bot_active_users', 'Currently active users')
queue_size = Gauge('bot_queue_size', 'Messages waiting in queue')
 
# Распределение времени обработки
request_duration = Histogram(
    'bot_request_duration_seconds',
    'Time spent processing request',
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0]  # Бакеты в секундах
)
 
# Декоратор для автоматического измерения времени
def track_time(func):
    async def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = await func(*args, **kwargs)
            messages_total.labels(command=func.__name__, status='success').inc()
            return result
        except Exception as e:
            messages_total.labels(command=func.__name__, status='error').inc()
            raise
        finally:
            duration = time.time() - start
            request_duration.observe(duration)
    return wrapper
 
@track_time
async def handle_start_command(message):
    active_users.inc()
    try:
        await process_message(message)
    finally:
        active_users.dec()
 
# Запускаем HTTP сервер для Prometheus
start_http_server(8000)  # Метрики доступны на :8000/metrics

import uuid
import aiohttp
from contextvars import ContextVar
 
trace_context = ContextVar('trace_id', default=None)
 
async def handle_webhook(request):
    # Генерим trace_id для нового запроса
    trace_id = str(uuid.uuid4())
    trace_context.set(trace_id)
    
    # Логируем с trace_id
    logger.info("webhook_received", trace_id=trace_id)
    
    # Обрабатываем запрос
    update = await request.json()
    await process_update(update)
 
async def call_payment_service(order_data):
    # Получаем trace_id из контекста
    trace_id = trace_context.get()
    
    async with aiohttp.ClientSession() as session:
        # Передаём trace_id следующему сервису
        headers = {'X-Trace-ID': trace_id}
        async with session.post(
            'http://payment-service/process',
            json=order_data,
            headers=headers
        ) as resp:
            logger.info(
                "payment_service_called",
                trace_id=trace_id,
                status=resp.status
            )
            return await resp.json()
 
# В payment service читаем заголовок
async def payment_handler(request):
    # Извлекаем trace_id из headers
    trace_id = request.headers.get('X-Trace-ID')
    trace_context.set(trace_id)
    
    logger.info("payment_processing", trace_id=trace_id)
    # Дальше обрабатываем с тем же trace_id

import sentry_sdk
from sentry_sdk.integrations.aiohttp import AioHttpIntegration
 
# Инициализация при старте бота
sentry_sdk.init(
    dsn="https://xxx@sentry.io/yyy",
    traces_sample_rate=0.1,  # 10% запросов для performance monitoring
    environment="production",
    release=f"bot@{VERSION}"
)
 
# Автоматический захват исключений
async def process_order(order_data):
    try:
        # Бизнес-логика
        result = await create_order(order_data)
    except Exception as e:
        # Exception автоматически уйдёт в Sentry
        # Добавляем контекст для более полной картины
        sentry_sdk.set_context("order", {
            "user_id": order_data.get('user_id'),
            "items_count": len(order_data.get('items', [])),
            "total": order_data.get('total')
        })
        
        # Добавляем breadcrumb для истории действий
        sentry_sdk.add_breadcrumb(
            category='order',
            message='Order creation failed',
            level='error'
        )
        
        raise  # Прокидываем дальше после логирования
 
# Для критичных мест явно шлём в Sentry даже обработанные ошибки
async def charge_payment(amount, user_id):
    try:
        result = await payment_gateway.charge(amount)
        return result
    except PaymentGatewayTimeout as e:
        # Обработали но хотим знать что таймауты случаются
        sentry_sdk.capture_exception(e)
        # Используем fallback
        return await retry_with_alternative_gateway(amount)

import os
 
# Никогда так
BOT_TOKEN = "1234567890:ABC-DEF1234ghIkl-zyx57W2v1u123ew11"
 
# Всегда так
BOT_TOKEN = os.getenv('BOT_TOKEN')
if not BOT_TOKEN:
    raise ValueError("BOT_TOKEN environment variable is required")
 
DATABASE_URL = os.getenv('DATABASE_URL', 'postgresql://localhost/bot_db')
REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')

# .env файл в корне проекта
BOT_TOKEN=1234567890:ABC-DEF1234ghIkl-zyx57W2v1u123ew11
DATABASE_URL=postgresql://user:password@localhost/bot_db
REDIS_URL=redis://localhost:6379/0
SENTRY_DSN=https://xxx@sentry.io/yyy
 
# Загрузка в Python коде
from dotenv import load_dotenv
import os
 
# Грузим переменные из .env перед использованием
load_dotenv()
 
BOT_TOKEN = os.getenv('BOT_TOKEN')

import boto3
from botocore.exceptions import ClientError
import json
from datetime import datetime, timedelta
 
class AWSSecretsManager:
    def __init__(self, region_name='us-east-1'):
        self.client = boto3.client('secretsmanager', region_name=region_name)
        self.cache = {}
        self.cache_ttl = timedelta(minutes=5)
    
    def get_secret(self, secret_name: str) -> dict:
        # Проверяем кеш
        if secret_name in self.cache:
            cached_at, value = self.cache[secret_name]
            if datetime.now() - cached_at < self.cache_ttl:
                return value
        
        try:
            # Получаем секрет из AWS
            response = self.client.get_secret_value(SecretId=secret_name)
            secret_value = json.loads(response['SecretString'])
            
            # Кешируем
            self.cache[secret_name] = (datetime.now(), secret_value)
            return secret_value
        except ClientError as e:
            if e.response['Error']['Code'] == 'ResourceNotFoundException':
                raise ValueError(f"Secret {secret_name} not found")
            raise
 
# Использование
secrets = AWSSecretsManager()
bot_config = secrets.get_secret('production/telegram-bot')
 
BOT_TOKEN = bot_config['bot_token']
DATABASE_URL = bot_config['database_url']

import os
 
ENVIRONMENT = os.getenv('ENVIRONMENT', 'development')
 
def get_secret_name(key: str) -> str:
    return f"{ENVIRONMENT}/telegram-bot/{key}"
 
# В зависимости от ENVIRONMENT подтянутся разные секреты
bot_token = secrets.get_secret(get_secret_name('bot_token'))

def get_bot_token():
    try:
        # Пробуем secrets manager сначала
        config = secrets.get_secret('production/telegram-bot')
        return config['bot_token']
    except Exception as e:
        logger.warning(f"Failed to load from secrets manager: {e}")
        # Fallback на environment variable
        token = os.getenv('BOT_TOKEN')
        if not token:
            raise RuntimeError("Cannot load bot token from any source")
        return token

import pytest
from services.promo import PromoService, PromoNotFound
 
@pytest.fixture
def promo_service():
    # Простой сервис без зависимостей от Telegram
    return PromoService()
 
def test_valid_promo_returns_discount(promo_service):
    result = promo_service.validate("SUMMER2024")
    assert result.is_valid
    assert result.discount == 0.20
 
def test_expired_promo_returns_error(promo_service):
    result = promo_service.validate("WINTER2023")
    assert not result.is_valid
    assert "expired" in result.error
 
def test_invalid_promo_raises_exception(promo_service):
    with pytest.raises(PromoNotFound):
        promo_service.validate("FAKE999")

from aiogram.types import User, Chat, Message
 
class MessageFactory:
    @staticmethod
    def create_message(
        text="/start",
        user_id=12345,
        chat_id=12345,
        username="testuser"
    ):
        user = User(
            id=user_id,
            is_bot=False,
            first_name="Test",
            username=username
        )
        chat = Chat(id=chat_id, type="private")
        
        return Message(
            message_id=1,
            date=datetime.now(),
            chat=chat,
            from_user=user,
            text=text
        )
 
# В тестах использование проще
def test_start_command():
    msg = MessageFactory.create_message("/start")
    # Дальше тест...

import pytest
import asyncpg
from alembic import command
from alembic.config import Config
 
@pytest.fixture(scope="session")
async def test_db():
    # Поднимаем тестовую БД
    pool = await asyncpg.create_pool(
        "postgresql://test:test@localhost:5433/test_db"
    )
    
    # Накатываем миграции
    alembic_cfg = Config("alembic.ini")
    command.upgrade(alembic_cfg, "head")
    
    yield pool
    
    # Чистим после тестов
    await pool.close()
 
@pytest.fixture
async def clean_db(test_db):
    # Перед каждым тестом чистим таблицы
    async with test_db.acquire() as conn:
        await conn.execute("TRUNCATE users, orders, products CASCADE")
    yield test_db

bot-aggregator/
├── src/
│   ├── core/                    # Ядро системы
│   │   ├── __init__.py
│   │   ├── config.py            # Конфигурация из env
│   │   ├── container.py         # DI контейнер
│   │   ├── database.py          # Пулы соединений
│   │   ├── event_bus.py         # Шина событий
│   │   └── exceptions.py        # Базовые исключения
│   │
│   ├── modules/                 # Бизнес-модули
│   │   ├── content/            # Управление контентом
│   │   │   ├── handlers.py     # Telegram handlers
│   │   │   ├── services.py     # Бизнес-логика
│   │   │   ├── repositories.py # Работа с БД
│   │   │   ├── models.py       # Domain модели
│   │   │   └── events.py       # События модуля
│   │   │
│   │   ├── channels/           # Управление каналами
│   │   │   ├── handlers.py
│   │   │   ├── services.py
│   │   │   ├── repositories.py
│   │   │   └── plugin.py       # Plugin interface
│   │   │
│   │   ├── analytics/          # Статистика и метрики
│   │   │   ├── handlers.py
│   │   │   ├── services.py
│   │   │   ├── collectors.py   # Сбор метрик
│   │   │   └── aggregators.py  # Агрегация данных
│   │   │
│   │   └── scheduler/          # Отложенные публикации
│   │       ├── tasks.py        # Celery задачи
│   │       ├── services.py
│   │       └── queue.py        # Priority queue
│   │
│   ├── infrastructure/          # Инфраструктурный слой
│   │   ├── telegram/           # Telegram API клиент
│   │   │   ├── client.py
│   │   │   └── rate_limiter.py
│   │   │
│   │   ├── cache/              # Кеширование
│   │   │   ├── redis_cache.py
│   │   │   └── strategies.py
│   │   │
│   │   └── monitoring/         # Метрики и логи
│   │       ├── metrics.py
│   │       └── logger.py
│   │
│   └── main.py                  # Точка входа
│
├── migrations/                  # Alembic миграции
│   └── versions/
│
├── tests/                       # Тесты
│   ├── unit/
│   ├── integration/
│   └── fixtures/
│
├── docker-compose.yml          # Локальное окружение
├── Dockerfile                  # Production образ
├── requirements.txt            # Зависимости
└── alembic.ini                # Конфиг миграций

# Presentation - handler знает только про Telegram типы
@router.message(Command("create_post"))
async def handle_create_post(
    message: Message,
    state: FSMContext,
    post_service: PostService  # Инжектится через DI
):
    # Только валидация и маршрутизация
    if not message.text or len(message.text) < 10:
        await message.answer("Пост слишком короткий")
        return
    
    try:
        # Вызываем service с простыми типами
        post = await post_service.create_post(
            author_id=message.from_user.id,
            content=message.text,
            channel_ids=[123, 456]  # Из FSM context
        )
        
        # Форматируем ответ для Telegram
        await message.answer(
            f"Пост #{post.id} создан и запланирован"
        )
    except InsufficientPermissions:
        await message.answer("У вас нет прав для публикации")
    except ValidationError as e:
        await message.answer(f"Ошибка валидации: {e}")
 
# Application - service содержит бизнес-логику
class PostService:
    def __init__(
        self,
        post_repo: PostRepository,
        channel_repo: ChannelRepository,
        event_bus: EventBus
    ):
        self._posts = post_repo
        self._channels = channel_repo
        self._events = event_bus
    
    async def create_post(
        self,
        author_id: int,
        content: str,
        channel_ids: list[int]
    ) -> Post:
        # Проверяем права - это бизнес-логика
        if not await self._has_permissions(author_id, channel_ids):
            raise InsufficientPermissions()
        
        # Валидация контента
        if len(content) > 4096:  # Telegram limit
            raise ValidationError("Content too long")
        
        # Создаём domain объект
        post = Post(
            author_id=author_id,
            content=content,
            status=PostStatus.DRAFT,
            created_at=datetime.now()
        )
        
        # Сохраняем через repository
        saved_post = await self._posts.create(post)
        
        # Связываем с каналами
        for channel_id in channel_ids:
            await self._posts.add_channel(saved_post.id, channel_id)
        
        # Публикуем событие для других модулей
        await self._events.publish(PostCreatedEvent(
            post_id=saved_post.id,
            author_id=author_id,
            channel_ids=channel_ids
        ))
        
        return saved_post