Пишем на Python'е сторожа для важных задач

Запустили бэкап или сборку? Теперь можно пойти пить кофе и молиться, чтобы всё прошло well. Или написать простого бота-напоминалку, который отправит вам в Telegram сообщение о результате.

🛠 Пример:
1. Создаем бота у [@BotFather](https://t.me/BotFather), получаем TOKEN.
2. Узнаем ID чата: проще всего написать боту [@userinfobot](https://t.me/userinfobot), он его вам скажет.
3. Скрипт-напоминалка (notify.py):

import requests

TOKEN = "YOUR_TELEGRAM_BOT_TOKEN"
CHAT_ID = "YOUR_CHAT_ID"
MESSAGE = "✅ Бэкап успешно завершен! Можно спать спокойно."

url = f"https://api.telegram.org/bot{TOKEN}/sendMessage"
params = {"chat_id": CHAT_ID, "text": MESSAGE}

requests.post(url, params=params).json()

4. Вставляем в скрипт бэкапа: python3 notify.py

Скрипт делает простой HTTP-запрос к Telegram API. Главное — не храните токен и chat_id в коде, если проект публичный! Используйте переменные окружения.

А вы куда получаете оповещения? Telegram, Slack, может быть, почта? И что мониторите? Расскажите о своем опыте автоматизации уведомлений!

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

🌐 @helcode | #python

Мониторим свое приложение: пишем кастомные метрики для Prometheus

Встроенных метрик системы часто недостаточно. Хочется знать, сколько пользователей онлайн, сколько запросов в секунду обрабатывает Ваше приложение, какая бизнес-логика выполняется. Давайте научим ваше приложение разговаривать с Prometheus!

🛠 Пример:

from prometheus_client import Counter, Gauge, start_http_server
import time

# Создаем метрики
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP Requests')
USERS_ONLINE = Gauge('users_online', 'Number of online users')

# В обработчике запроса
def handle_request():
    REQUEST_COUNT.inc()  # Увеличиваем счетчик
    USERS_ONLINE.set(get_online_users())  # Устанавливаем значение

# Запускаем HTTP сервер для метрик
start_http_server(8000)

➤ Counter - только увеличивается (запросы, ошибки)
➤ Gauge - может увеличиваться и уменьшаться (пользователи онлайн, температура)
➤ Histogram - распределение значений (время ответа)

🌐 @helcode | #python

Кешируем результаты API-запросов в Redis

Ваш скрипт делает запросы к медленному API, которое отвечает одними и теми же данными. Давайте сохраним результат в Redis и будем использовать его вместо повторных запросов!

🛠 Пример:

import redis
import requests
import json
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def get_cached_data(user_id):
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    
    if cached:
        print("Данные из кеша!")
        return json.loads(cached)
    
    # Если нет в кеше - делаем запрос к API
    print("Делаем запрос к API...")
    response = requests.get(f"https://api.example.com/users/{user_id}")
    data = response.json()
    
    # Сохраняем в кеш на 5 минут
    r.setex(cache_key, 300, json.dumps(data))
    return data

# Использование
user_data = get_cached_data(123)

setex устанавливает значение с временем жизни. Через 5 минут ключ автоматически удалится.

А вы кешируете запросы к внешним API? Какие библиотеки или подходы используете?

Кеширование API-запросов - это как записывать сериал на флешку: не нужно ждать загрузки каждый раз, когда хотите пересмотреть.

P.S. Если Вам интересно, как мы настраивали кеширование в нашем боте и с чем столкнулись на практике - пишите об этом в комментариях! Спойлер: ошибок было действительно много...

🌐 @helcode | #python

Пишем простую систему обработки задач с Redis RQ

Нужен простой способ выполнять фоновые задачи без настройки сложных брокеров? Redis RQ (Redis Queue) - минималистичное и эффективное решение для Python.

Код:

# worker.py
import requests
from rq import Queue
from redis import Redis
from worker import count_words

# Подключаемся к Redis
redis_conn = Redis()
q = Queue(connection=redis_conn)

# Отправляем задачу в очередь
job = q.enqueue(count_words, 'https://example.com')
print(f"Задача {job.id} добавлена в очередь!")

# Запускаем воркер в отдельном терминале:
# rq worker

# tasks.py
import requests

def count_words(url):
    response = requests.get(url)
    word_count = len(response.text.split())
    print(f"На странице {url} слов: {word_count}")
    return word_count

RQ использует Redis как брокер. Воркеры постоянно слушают очередь и выполняют задачи по мере их поступления.

А Вы пробовали RQ или другие простые системы очередей? Как впечатления?

Redis RQ - это как нанять стажера на подработку: отдаешь задачу и забываешь, а он тихо ее выполняет где-то в фоне.

🌐 @helcode | #python

Пишем свою первую Lambda функцию на Python

Хотите попробовать serverless, но кажется сложным? Давайте создадим простейшую Lambda функцию, которая возвращает "Hello, World!".

🛠 Пример:

import json

def lambda_handler(event, context):
    # event содержит данные запроса
    name = event.get('name', 'World')
    
    return {
        'statusCode': 200,
        'body': json.dumps({
            'message': f'Hello, {name}!'
        })
    }

Как развернуть:

# Упаковываем код в zip
zip function.zip lambda_function.py

# Создаем функцию в AWS
aws lambda create-function --function-name hello-world \
--runtime python3.9 --role arn:aws:iam::123456789012:role/lambda-role \
--handler lambda_function.lambda_handler --zip-file fileb://function.zip

# Тестируем
aws lambda invoke --function-name hello-world \
--payload '{"name": "Digital Automation Lab"}' output.txt

Функция получает event (данные запроса) и context (информация о выполнении), а возвращает ответ в определенном формате.

А вы уже пробовали Lambda? Какие функции написали?

Написание Lambda функции - это как отправка письма по почте: написал, отправил, а почтальон (AWS) уже знает, куда его доставить и как обработать.

🌐 @helcode | #python