Отправляем уведомления в Telegram на Python

В современной backend-разработке Telegram остается стандартом для мгновенных алертов, мониторинга состояния ИИ-агентов и уведомлений о критических событиях инфраструктуры

1. Установка библиотеки
Одним из хороший вариантов является httpx благодаря поддержке HTTP/2 и возможности легкого перехода на асинхронность

pip install httpx

2. Импорт и конфигурация
Импортируем клиент и используем переменные окружения для защиты учетных данных

import httpx
import os

TOKEN = os.getenv("TG_BOT_TOKEN")
CHAT_ID = os.getenv("TG_CHAT_ID")

3. Отправка сообщения с валидацией
Формируем запрос

def send_telegram_notification(text: str):
    url = f"https://api.telegram.org/bot{TOKEN}/sendMessage"
    payload = {"chat_id": CHAT_ID, "text": text}
    
    with httpx.Client(http2=True, timeout=10.0) as client:
        r = client.post(url, json=payload)
        r.raise_for_status()
        
        resp = r.json()
        # Важно учесть: Telegram возвращает HTTP 200 даже при ошибках логики бота
        if not resp.get("ok"):
            raise RuntimeError(f"Telegram API Error: {resp}")
        return resp

send_telegram_notification("Система мониторинга: все в норме :)")

⭐️ Пост Guru Python:
Сергей Филичкин

🚀 В январе перед нами стояли три ключевые задачи.

1. Миграция на Next.js
Мы перевели публичную часть проекта на Next.js, а дашборд оставили на React.js. Основные работы почти завершены — сейчас занимаемся тестированием и финальными правками.
В процессе столкнулись с рядом инфраструктурных вызовов: текущих ресурсов сервера оказалось недостаточно под новые проекты, поэтому пришлось масштабироваться. Впереди также большая работа по настройке SEO.

2. Сервис для лайвкодинга
Мы разрабатываем сервис лайвкодинга — аналог LeetCode с задачами российских компаний с реальных собеседований. У проекта большие планы:

— проведение контестов,
— обучающие модули,
— курс по алгоритмам,
— ИИ-помощник с построчным разбором решений,
— челленджи и соревнования между пользователями (пока держим в секрете).

Базовая версия сервиса уже готова и проходит тестирование. Совсем скоро — релиз.

3. Сервис уведомлений
Мы начали разработку сервиса уведомлений, который объединит всю нашу экосистему: платформу, Telegram-каналы и бота.

В Telegram будут публиковаться:

— новые собеседования,
— вопросы и задачи,
— статьи и подборки,
— анонсы,
— вакансии,
— статистика с HH.

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

Дополнительно
В работе также находится сервис тестов с вариантами ответов и системой разделения вопросов по темам — для будущего роадмапа. Пока эти направления развиваются в «лайт»-режиме.

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

P.S. Новогодние скидки продолжают действовать до запуска сервиса лайвкодинга. После релиза цены увеличатся в 2 раза — сейчас самое выгодное время подключиться и поддержать нас.

👉 Членство YeaHub

Впереди много интересного. Спасибо, что вы с нами.✊🏻🚀💪

Потоковая обработка данных. Как еще можно применять генераторы в Python

Генераторы стали основой для работы с непрерывными потоками информации: live-метрики, события IoT, стримы логов и real-time аналитика

Например генератор потока данных от сенсора

import time
import random

def sensor_stream(sensor_id):
    """Поток показаний сенсора"""
    value = 50.0
    while True:
        value += 1  # Новое показание
        yield {
            'sensor': sensor_id,
            'value': value,
            'timestamp': time.time()
        }
        time.sleep(1)  # Задержка между измерениями

Фильтрация + агрегация на лету

def process_stream(stream, max_value=100):
    """Фильтрует аномалии и считает среднее за последние 5 значений"""
    buffer = []
    for point in stream:
        if 0 < point['value'] < max_value:  # Фильтрация
            buffer.append(point['value'])
            buffer = buffer[-5:]  # Последние 5 значений
            
            point['avg'] = sum(buffer) / len(buffer)
            yield point

Применение в мониторинге систем

# Создаём поток данных
stream = sensor_stream('server_temp')
processed = process_stream(stream)

# Мониторим
for point in processed:
    print(f"{point['sensor']}: {point['value']}°C (avg: {point['avg']:.1f})")
    
    if point['avg'] > 30:
        print("⚠️ Внимание")

Ключевые преимущества такого подхода:
1. Композируемость — генераторы легко объединяются в цепочки
2. Ресурсная эффективность — не храним полные массивы данных
3. Раннее обнаружение проблем — многие проблемы видно сразу, а не после обработки всего массива

⚡️ Генераторы позволяют обрабатывать данные в момент их появления, что критично для систем, в которых нужно отслеживать данные в реальном времени

⭐️ Пост Guru Python:
Сергей Филичкин

Когда действительно нужны итераторы в python

И первое с чего нужно начать — итераторы нужны не ради экономия памяти, а они нужны ради контроля над чтением данных

Самый простой пример:

class DataStream:
    def __init__(self, source):
        self.source = source
        self.pos = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.pos >= len(self.source):
            raise StopIteration
        value = self.source[self.pos]
        self.pos += 1
        return value

Такой объект читает данные по одному элементу и запоминает позицию

Т.е. состояние хранится внутри и повторно пройтись по данным нельзя

stream = DataStream([1, 2, 3])

list(stream)  # [1, 2, 3]
list(stream)  # []

По сути если объект можно читать сколько угодно раз — это скорее коллекция
А если можно прочитать только один раз — это поток

Итератор — это одноразовый поток

В целом можно сделать и через генератор:

def stream(data):
    for x in data:
        yield x

Но так стоит делать в следующих ситуациях:
- нет необходимости возвращаться назад
- нет внешних эффектов
- не нужно управление процессом чтения

Но как только появится задача вроде "продолжить с места последнего падения" или "повторить последний шаг" — с генератором может быть больше проблем, чем пользы

Как можно сохранить позицию, например:

class CheckpointStream(DataStream):
    def checkpoint(self):
        return self.pos

    def restore(self, pos):
        self.pos = pos

stream = CheckpointStream([100, 200, 300])

print(next(stream))  # 100
saved = stream.checkpoint()

print(next(stream))  # 200
stream.restore(saved)

print(next(stream))  # снова 200

Так работают пакетные обработки, конвейеры данных и повторные запуски после ошибок


А асинхронный вариант может выглядеть так:

class AsyncStream:
    def __aiter__(self):
        return self

    async def __anext__(self):
        ...

# Дальше обычный цикл:
async for item in stream:
    ...

Если подытожить:
Генератор — удобный синтаксис
Итератор — инструмент управления

Если контроль над данными и позициями не нужен — не усложняйте код, можете просто использовать генератор


⭐️ Пост Guru Python:
Сергей Филичкин