Скрипты-самоубийцы: как автоматически остановить скрипт, если он сломался

Ваш скрипт завис и висит сутками, съедая ресурсы? Научите его вовремя уходить с достоинством!

Скрипты, которые работают с внешними ресурсами (сеть, БД, API), могут внезапно зависнуть на неопределенное время. Это съедает память, блокирует другие процессы и портит картину мониторинга. Нужен встроенный механизм самоликвидации.

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

Способ 1: таймаут на отдельную функцию (декоратор)

import signal
from functools import wraps

class TimeoutException(Exception):
    pass

def timeout(seconds=10):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Устанавливаем обработчик сигнала ALARM
            def handle_timeout(signum, frame):
                raise TimeoutException("Функция выполнена дольше {seconds} секунд!")
            
            old_handler = signal.signal(signal.SIGALRM, handle_timeout)
            signal.alarm(seconds)  # Заводим будильник
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)  # Отключаем будильник
                signal.signal(signal.SIGALRM, old_handler)  # Восстанавливаем старый обработчик
            return result
        return wrapper
    return decorator

# Использование
@timeout(5)  # Максимум 5 секунд на выполнение
def call_slow_api():
    import time
    time.sleep(10)  # Имитация долгого запроса
    return "success"

try:
    call_slow_api()
except TimeoutException:
    print("Функция прервана по таймауту!")

➤ Сигнал SIGALRM: посылается ядром Linux по истечении времени, заданного signal.alarm().
➤ Декоратор @timeout: позволяет обернуть любую функцию, установив для нее лимит выполнения.
➤ Важно: этот метод работает только в главном потоке и не для всех операций (например, некоторые системные вызовы могут быть не прерваны).

Сталкивались ли вы с зависанием скриптов?

Научить скрипт самоликвидироваться — это как дать секретному агенту капсулу с ядом. Жестко, но эффективно для миссии.

🌐 @helcode | #python

Кэширование для ленивых: ускоряем скрипты в 100 раз без усилий

Ваш скрипт 100 раз ходит за одними и теми же данными? Пора дать ему память!

🏋️ Многие скрипты делают одни и те же «тяжелые» операции при каждом запуске: чтение больших файлов конфигурации, запросы к медленным API, сложные вычисления. Кэширование результатов между запусками — простой способ ускорить их в разы.

Используем декоратор @cache из стандартной библиотеки Python (появился в 3.9+).

Кэширование результатов тяжелой функции:

from functools import cache
import time

@cache  # Волшебная строка, которая все кэширует
def get_weather(city):
    # Имитация медленного HTTP-запроса к API погоды
    time.sleep(2)
    # Здесь реальный запрос: requests.get(f"http://weather.com/{city}").json()
    return f"Погода в {city}: +25°C"  # Возвращаем заглушку

# Первый вызов — медленный (2 секунды)
print(get_weather("Москва"))  

# Все последующие вызовы с ТЕМ ЖЕ аргументом — мгновенные!
print(get_weather("Москва"))  # Результат взят из кэша
print(get_weather("Москва"))  # Снова из кэша

# Вызов с новым аргументом — снова медленный (2 секунды)
print(get_weather("Санкт-Петербург"))  

# Кэш хранится ТОЛЬКО в течение работы программы.
# Для кэша между запусками нужны другие решения (redis, diskcache)

➤ @cache (или @lru_cache из functools для более старых версий) запоминает аргументы, с которыми вызывалась функция, и ее возвращаемое значение.
➤ При повторном вызове с теми же аргументами функция не выполняется, а результат возвращается из памяти.
➤ Важно: aункция должна быть чистой (pure) - возвращать одинаковый результат для одинаковых аргументов без side effects.

Где вы используете кэширование?

Добавить @cache - это как поставить термос с кофе рядом с кроватью. Проснулся, выпил, уже бодрый - не надо идти на кухню и варить заново.

🌐 @helcode | #python

Как избежать хаоса: конфигурация скриптов в YAML

Ваши скрипты обрастают десятками параметров в коде? Пора вынести их в конфиг!

«Жесткие» параметры внутри кода (пути, адреса, пороги срабатывания) - это зло. При каждом изменении нужно лезть в код, рискуя что-то сломать. Правильное решение - выносить конфигурацию в отдельные файлы.

Используем YAML - человекочитаемый формат для конфигов.

Пример файла config.yaml:

database:
  host: "localhost"
  port: 5432
  name: "my_app"
  user: "admin"

logging:
  level: "INFO"
  file: "/var/log/my_app.log"

api:
  endpoint: "https://api.example.com/v1"
  timeout_seconds: 30
  retry_attempts: 3

features:
  enable_telemetry: false
  cache_size: 1000

Загрузка конфига в Python:

import yaml  # pip install PyYAML

with open("config.yaml", "r") as f:
    config = yaml.safe_load(f)

# Теперь доступ к параметрам прост и безопасен
db_host = config['database']['host']
timeout = config['api']['timeout_seconds']

➤ YAML поддерживает сложные структуры (словари, списки), комментарии и более читаем, чем JSON.
➤ Отделение кода от конфигурации - ключевая практика DevOps. Это позволяет использовать один и тот же код в разных окружениях (dev, prod) без изменений.
➤ Версионирование: rонфиг-файлы можно хранить в git отдельно от кода с разными ветками для разных сред.

Как вы храните конфигурацию?

Переход на YAML-конфиги - это как перейти от управления государством через указы, написанные на салфетке, к полноценной конституции. Порядок и предсказуемость!

🌐 @helcode | #yaml #python

Автоматическое тестирование скриптов: оно вам нужно?

Вы запускаете скрипт и молитесь, чтобы он не сломал прод? Пора подружиться с тестами!

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

🛠 Пишем простой unit-тест для функции из скрипта с помощью pytest.

Допустим, у нас есть функция в script.py:

# script.py
def process_data(input_data):
    """Удаляет все числа меньше 10 из списка."""
    if not isinstance(input_data, list):
        raise ValueError("Input must be a list!")
    return [x for x in input_data if x >= 10]

Пишем тест в файле test_script.py:

# test_script.py
import pytest
from script import process_data

# Стандартный тест на корректные данные
def test_process_data_positive():
    assert process_data([1, 10, 20, 5]) == [10, 20]

# Тест на пограничное условие
def test_process_data_boundary():
    assert process_data([9, 10]) == [10]

# Тест на некорректный ввод
def test_process_data_negative():
    with pytest.raises(ValueError):
        process_data("not a list")

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

pytest -v test_script.py

➤ Unit-тесты проверяют отдельные функции («юниты») изолированно от остального кода.
➤ pytest - стандарт де-факто для тестирования в Python: простой синтаксис, информативные сообщения об ошибках.
➤ Тесты помогают без страха рефакторить код и добавлять новую функциональность, не ломая старое.

Вы пишете тесты для своих скриптов автоматизации?

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

🌐 @helcode | #python

JQ: ваш карманный ниндзя для боя с JSON в терминале

Контекст: API сыпят на нас тоннами JSON. Ручной парсинг глазами — путь в никуда. Команда jq превращает эту кашу в структурированные данные, с которыми можно работать прямо в bash.

Код/Пример:
Допустим, у нас ответ от API такой: {"users": [{"name": "Alice", "id": 1}, {"name": "Bob", "id": 2}]}

➤ Достать всех юзеров: curl ... | jq '.users'
➤ Только имена: curl ... | jq '.users[].name' -> "Alice", "Bob"
➤ Имя первого юзера: curl ... | jq '.users[0].name' -> "Alice"
➤ Преобразовать в массив имен: curl ... | jq '[.users[].name]' -> ["Alice", "Bob"]

Ключ к успеху - точка (.) означает текущий объект, а комбинации фильтров [] (для массивов) и .key (для полей) творят "магию".

Какой ваш самый сложный или изящный jq-однострочник?

jq для JSON — это как световой меч для джедая: без него можно обойтись, но с ним ты чувствуешь силу.

🌐 @helcode | #python

Пишем на 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