Helcode | Хелкод | Скрипты и автоматизация
993 subscribers
52 photos
34 links
☎️ Контакты для связи: @helcodeadm
Download Telegram
Твой Python-скрипт вырос и хочет отдельной команды в терминале. Как ему это организовать? ☕️

Мы все начинаем с простого скрипта super_script.py. Но потом к нему добавляются аргументы, конфиги, логирование... И вот ты уже не запускаешь скрипт, а выполняешь сложный ритуал: python /path/to/script.py --verbose --output /home/user/data.json. Пора дать ему собственное имя и сделать полноценным CLI-инструментом!

Всё, что нужно, — это argparse из коробки Python. Смотрим, как превратить хаос в элегантную команду:

# file: data_processor.py
import argparse

def process_data(input_file, output_file, verbose=False):
# Здесь твоя магия обработки данных
if verbose:
print(f"Обрабатываем {input_file} и сохраняем в {output_file}")
return True

if __name__ == "__main__":
# Создаем парсер
parser = argparse.ArgumentParser(description='Крутой инструмент для обработки данных.')

# Добавляем аргументы
parser.add_argument('input', help='Путь до входного файла')
parser.add_argument('-o', '--output', required=True, help='Путь до выходного файла')
parser.add_argument('-v', '--verbose', action='store_true', help='Вербальный режим')

# Парсим аргументы
args = parser.parse_args()

# Передаем в функцию
process_data(args.input, args.output, args.verbose)


1. ArgumentParser — это наш главный конструктор команд.
2. add_argument определяет, какие флаги и аргументы мы принимаем. required=True для обязательных, action='store_true' для булевых флагов.
3. parse_args() превращает sys.argv в удобный объект с точками (args.input).

Следующий уровень: после этого можно установить скрипт как пакет через pip install -e . (прописав в setup.py) и твоя команда data_processor будет доступна из любой директории!

А какой у вас любимый инструмент для создания CLI? 🤔

P.S. Если твой скрипт после этого начнет требовать зарплату в виде GPU — мы не виноваты.


🌐 @helcode | #python
👍2🤔1
Как заставить старый скрипт работать в 10 раз быстрее

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

👁‍🗨 Мы все их видели: скрипты, написанные на коленке, которые со временем стали критически важными. Они работают, но отжирают уйму времени и ресурсов. Часто проблема не в сложности задачи, а в неоптимальных подходах.

👉Взгляните на этот кусок кода. Часто проблема в циклах внутри циклов, где можно применить списковые включения или функции из itertools.

Было (медленно):

results = []
for item in huge_list:
if some_condition(item):
for subitem in item.subitems:
if another_condition(subitem):
results.append(process(subitem))


Стало (быстро):

results = [process(subitem) for item in huge_list if some_condition(item) for subitem in item.subitems if another_condition(subitem)]
# Или еще лучше: использовать генератор для экономии памяти!
results_gen = (process(subitem) for item in huge_list ... )


Списковые включения в Python оптимизированы на уровне C и работают значительно быстрее классических for`-циклов с .append()`. А использование генераторов (`()`) вместо списков (`[]`) позволяет не загружать все данные в память сразу, что еще ускоряет работу с большими объемами данных.

А вы как ускоряли своих "динозавров"? Поделитесь самым крутым оптимизационным хаком в комментариях!

Это как в «Форсаже»: иногда не нужен новый двигатель, а нужно просто правильно настроить карбюратор и сбросить лишний вес.


🌐 @helcode | #python
👍3
Как автоматически генерировать сложные пароли для всей команды

Ваш пароль "qwerty123" прекрасен... если вы хотите подарить хакеру новый автомобиль. Давайте генерировать лучше!

Слабые пароли - главная дыра в безопасности. А придумывать и хранить десятки сложных паролей для разных сервисов не может даже человек с памятью как у Шерлока. Доверьте это машине.

Однострочник на Python в помощь всей команде:
import secrets
import string

alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
password = ''.join(secrets.choice(alphabet) for i in range(16))
print(password) # Пример: 'xT5&!k8@qL9#vR2$'


Мы используем модуль secrets, который криптографически безопасен (в отличие от random). string.ascii_letters + string.digits - это наш алфавит (буквы и цифры). Добавляем спецсимволы для сложности. Длина 16 символов - отличный старт.

А как вы управляете паролями в своей команде?

🌐 @helcode | #python
👍2
Автоматизируйте свои чат-боты: хватит отвечать на одни и те же вопросы!

Ваша команда снова спрашивает, «каой пароль от тестовой БД?».

🤖Пора научить бота делать вашу работу за вас.

В любой команде есть рутинные вопросы, которые отвлекают от работы. «Где документация?», «Как запустить проект?», «Кто дежурный?». Вместо того чтобы в сотый раз копировать один и тот же ответ, можно написать простого бота для Telegram или Slack.

Вот пример на Python с библиотекой python-telegram-bot для обработки частого вопроса.
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
# Бот отвечает на сообщение с ключевым словом
await update.message.reply_text(
"Коллега, привет! Вот частые ответы:\n"
"• Пароль от БД: 'supersecret'\n"
"• Документация: https://wiki.our-company.com/\n"
"• Как запустить: см. файл README.md в корне проекта"
)

if __name__ == '__main__':
app = Application.builder().token("YOUR_TOKEN").build()г
app.add_handler(MessageHandler(filters.TEXT & filters.Regex(r'(?i)(пароль|документация|как запустить)'), help_command))
app.run_polling()


Мы используем MessageHandler с фильтром filters.Regex(). Он ловит любое текстовое сообщение, в котором есть слова «пароль», «документация» или «как запустить» (регистр неважен благодаря `(?i)`), и автоматически отвечает заготовленным текстом.

Какой самый частый и раздражающий вопрос вы получаете в чате?

Это как поставить на частый вопрос автоответчик «я в отпуске», только Вы не в отпуске, а просто пьете кофе, пока бот работает.


🌐 @helcode | #python
👍2
Хватит хардкодить пароли в скриптах! Где и как хранить секреты

password = "12345" 
в коде — это как оставить ключи от квартиры в двери.

API-токены, пароли от БД, ключи шифрования. Они не должны лежать в открытом виде в коде, который вы заливаете на GitHub. Но как тогда скрипт получит к ним доступ? Решение есть.

Используем переменные окружения и файлы.

Способ 1: переменные окружения (просто и достаточно для многих случаев)

Задаем переменную в сессии или в файле ~/.bashrc / ~/.zshrc:export

DB_PASSWORD="my_super_secret_password_123"


И используем в Python-скрипте:

import os
db_pass = os.environ.get('DB_PASSWORD')
if not db_pass:
raise ValueError("Не задан пароль от БД!")


Способ 2: Файлы с секретами (более безопасно)

Создаем файл, который не будет попадать в git:

# .env file (добавляем в .gitignore!)
DB_HOST=localhost
DB_USER=admin
DB_PASSWORD=secret

Читаем его в коде с помощью библиотеки python-dotenv:

pip install python-dotenv
from dotenv import load_dotenv
load_dotenv() # Загружает переменные из файла .env в os.environ
db_pass = os.getenv('DB_PASSWORD')

Немного теории:

- Переменные окружения — это изолированное пространство памяти процесса. Их не видно в списке запущенных процессов (в отличие от аргументов командной строки).
- .env файлы — это удобный способ локальной разработки. На продакшене (сервере) эти же переменные задаются через конфигурацию systemd, Docker или панели управления хостингом.


Где вы храните свои секреты?
Признавайтесь, мы не осудим


Хардкодить пароль в коде — это как написать пин-код от карты на ней же шариковой ручкой. Удобно, но недальновидно.


🌐 @helcode | #bash #python
👍3🤣1
multiprocessing или threading? Вопрос на миллион потоков... и процессов!

Ваш скрипт тормозит и вы хотите распараллелить все на свете.

🤔 Но что выбрать? Потоки? Процессы?

Ответ кроется в трех буквах — GIL (Global Interpreter Lock).

☕️ Это такой злой бариста в кофейне Python, который не дает нескольким потокам одновременно молоть зёрна (выполнять Python-код). Но у каждого процесса своя кофейня (и свой GIL)!

Код-пример (CPU-bound):
# Много вычислений -> multiprocessing
import multiprocessing
import math

def calculate_factorial(n):
return math.factorial(n)

if __name__ == '__main__':
numbers = [100000, 200000, 300000] * 5 # Тяжелая работа
with multiprocessing.Pool() as pool:
results = pool.map(calculate_factorial, numbers) # 🚀 Запуск на всех ядрах!


Код-пример (I/O-bound):
# Много ожидания -> threading
import threading
import requests

def check_site(url):
response = requests.get(url)
return response.status_code

urls = ['https://google.com', 'https://yandex.ru', ...] * 10
threads = []
for url in urls:
thread = threading.Thread(target=check_site, args=(url,))
thread.start()
threads.append(thread)

for thread in threads:
thread.join() # 🧵 Ждем все потоки


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

А вы чаще сталкиваетесь с I/O или CPU-bound задачами? И какой ваш любимый способ паралеллить?

P.S. GIL — это не баг, это фича, которая не дает твоему коду сломать интерпретатор. Но иногда так хочется его пристрелить... (шучу, не шучу).


🌐 @helcode | #python
Ваши логи выглядят как поток сознания писателя-авангардиста? Пора навести порядок!

📖 Читать обычные текстовые логи, где вперемешку идут отладочные сообщения, ошибки и информационные записи, - это как искать иголку в стоге сена. А если нужно автоматически анализировать логи? На помощь приходят структурированные логи в формате JSON.

Используем встроенный модуль logging в Python, но с кастомным форматтером.

Настройка структурированного логирования:
import logging
import json
from datetime import datetime

class StructuredMessage:
def __init__(self, message, **kwargs):
self.message = message
self.kwargs = kwargs

def __str__(self):
# Создаем словарь с основным сообщением и дополнительными полями
log_data = {
'timestamp': datetime.utcnow().isoformat() + 'Z',
'level': logging.getLevelName(logging.getLogger().getEffectiveLevel()),
'message': self.message,
**self.kwargs # Добавляем все переданные ключевые аргументы
}
return json.dumps(log_data)

# Настраиваем логирование
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
# Не нужно форматировать, тк наши сообщения уже строки JSON
handler.setFormatter(logging.Formatter('%(message)s'))
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Использование
logger.info(StructuredMessage("User logged in", user_id=123, ip="192.168.1.1", success=True))
logger.error(StructuredMessage("DB connection failed", attempt=3, timeout=30))


Вместо неструктурированной строки мы получаем JSON-объект:
{"timestamp": "2023-10-26T10:30:15.123456Z", "level": "INFO", "message": "User logged in", "user_id": 123, "ip": "192.168.1.1", "success": true}


Преимущества:
- Легко парсить автоматизированными системами (ELK Stack, Loki, CloudWatch).
- Можно добавлять любые метаданные к каждому событию (user_id, request_id, execution_time).
- Упрощен поиск и фильтрация по конкретным полям.

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

Переход на структурированные логи - это как перейти от расшифровки древних свитков к аккуратной Excel-таблице. Информация та же, а работать в разы удобнее.


🌐 @helcode | #python
👍3
Как заставить скрипты общаться друг с другом: не только HTTP

Ваши скрипты живут в вакууме и не общаются? Пора познакомить их друг с другом! 🤝

Часто одна автоматизированная задача должна запускать другую или передавать ей данные. Делать это через запись в файл и постоянный опрос - неэлегантно. Есть более эффективные способы межпроцессного взаимодействия (IPC).

👀 Рассмотрим несколько способов, кроме очевидного HTTP.

Способ 1: очереди сообщений (Message Queues) на примере Redis
# Скрипт-отправитель (producer.py)
import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)
task = {'user_id': 456, 'action': 'generate_report'}
r.lpush('task_queue', json.dumps(task)) # Кладем задачу в очередь


# Скрипт-получатель (consumer.py)
import redis
import json
import time

r = redis.Redis(host='localhost', port=6379, db=0)
while True:
# Блокирующее ожидание задачи из очереди (ждем 30 сек)
_, task_json = r.brpop('task_queue', timeout=30)
if task_json:
task = json.loads(task_json)
print(f"Обрабатываю задачу: {task}")
# ... обработка ...
time.sleep(1)


Способ 2: Named Pipes (именованные каналы) в Linux/macOS
# Создаем канал
mkfifo my_pipe

# Скрипт-читатель (зависнет, пока не появятся данные)
cat my_pipe

# Скрипт-писатель
echo "Hello from another script!" > my_pipe


* Очереди сообщений идеальны для построения распределенных систем. Гарантируют доставку, позволяют балансировать нагрузку между несколькими "воркерами".
* Именованные каналы - очень простой способ одноразовой передачи данных между процессами на одной машине. Работает на уровне файловой системы.

Как ваши скрипты общаются между собой?

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


🌐 @helcode | #python #bash
👍4
Тайм-ауты и повторные попытки: делаем ваши скрипты устойчивыми к сбоям

Ваш скрипт падает от малейшей ошибки сети? Пора научить его быть настойчивее!

В реальном мире сеть ненадежна, сервисы падают, а ответы задерживаются. Наивный скрипт, который не учитывает этого, обречен на частые failures. Устойчивость (resilience) - ключевое свойство для продакшн-автоматизации.

Паттерн "Retry with backoff" (Повтор с экспоненциальной задержкой).

Реализация на Python с помощью библиотеки tenacity:
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests

@retry(
stop=stop_after_attempt(5), # Остановиться после 5 попыток
wait=wait_exponential(multiplier=1, min=2, max=30), # Экспоненциальная задержка: 2s, 4s, 8s... макс. 30s
retry=retry_if_exception_type(requests.exceptions.RequestException) # Повторять только при ошибках сети
)
def call_unreliable_api(url):
response = requests.get(url, timeout=3)
response.raise_for_status() # Выбросит исключение при HTTP-ошибке
return response.json()

# Использование
try:
data = call_unreliable_api("https://api.example.com/shaky-endpoint")
print("Успех!", data)
except Exception as e:
print(f"Не удалось после всех попыток: {e}")


wait_exponential: увеличивает задержку между попытками по экспоненте. Это вежливо по отношению к упавшему серверу (не DDoSим его) и дает ему время на восстановление.
stop_after_attempt: защита от бесконечных петель. Всегда должен быть лимит.
retry_if_exception_type: важно повторять только те ошибки, которые имеют смысл retry'ить (проблемы сети, тайм-ауты, 5xx ошибки сервера). Ошибку 404 (Not Found) retry'ить бессмысленно.

С какими сбоями чаще всего сталкиваются ваши скрипты?

Добавление retry-логики - это как дать скрипту несколько дополнительных жизней в видеоигре. Он может проиграть первый раунд, но в итоге все равно победит. 🎮


🌐 @helcode | #python
🤔2
Как избежать race condition: локи для идущих в ногу скриптов

🚪 Два скрипта заходят в одну папку... Звучит как начало плохой IT-шутки и большой проблемы!

Если один и тот же критический ресурс (файл, запись в БД, внешний API) могут одновременно использовать несколько экземпляров вашего скрипта - возникает состояние гонки (race condition). Результат может быть непредсказуемым: поврежденные данные, двойное списание средств, бессмысленная нагрузка.

Используем файловые блокировки (file locks). В Python для этого есть удобная библиотека filelock.

Пример: гарантируем, что только один экземпляр скрипта будет выполнять критическую операцию
from filelock import FileLock
import time

# Создаем lock-файл. Его наличие будет сигналом, что ресурс занят.
lock_path = "/tmp/my_script.lock"

# Используем менеджер контекста для гарантированного освобождения блокировки
with FileLock(lock_path, timeout=10): # Ждем 10 секунд, если lock занят
print("🔒 Lock acquired! Начинаю критическую операцию...")
# Здесь работа с общим ресурсом
time.sleep(5) # Имитация долгой работы
print(" Операция завершена. Lock released.")

# После выхода из блока 'with' блокировка автоматически снимается


➤ FileLock создает одноименный lock-файл. Попытка захватить lock, который уже у кого-то есть, будет блокировать выполнение (timeout) или вызовет исключение.
➤ Менеджер контекста (with) гарантирует, что блокировка будет снята даже если внутри блока произойдет ошибка.
➤ Это работает и между разными скриптами на разных языках, если они соблюдают одинаковый protocol (проверяют наличие одного и того же lock-файла).

В ваших скриптах есть критические секции? Как вы их защищаете?

Использование lock-файлов - это как повесить табличку «не беспокоить» на дверь офиса. Вежливо и предотвращает неловкие ситуации.


🌐 @helcode | #python
👍21
Парсим сайты, которые не хотят, чтобы их парсили: обход блокировок

Современные сайты защищаются от парсинга: капчи, блокировка по IP, анализ поведения браузера. Наивный requests.get() уже не работает. Нужно маскироваться под человека.

Используем Selenium WebDriver с Stealth-плагином и ротируем User-Agent.

Настройка "стелс"-браузера на Python:
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium_stealth import stealth
import random

# Создаем опции Chrome
options = Options()
options.add_argument("--headless=new") # Запуск в фоне
options.add_argument(f"--user-agent={random.choice(user_agents)}") # Рандомный UA
options.add_experimental_option("excludeSwitches", ["enable-automation"])
options.add_experimental_option('useAutomationExtension', False)

driver = webdriver.Chrome(options=options)

# Применяем stealth-плагин для сокрытия автоматизации
stealth(driver,
languages=["en-US", "en"],
vendor="Google Inc.",
platform="Win32",
webgl_vendor="Intel Inc.",
renderer="Intel Iris OpenGL Engine",
fix_hairline=True,
)

driver.get("https://hard-to-parse-site.com")
# Дальше работаем как с обычным браузером


➤ Ротация User-Agent: смена «отпечатка» браузера делает запросы менее похожими друг на друга.
➤ Selenium Stealth: плагин скрывает следы автоматизации, которые ищут анти-бот системы (например, наличие navigator.webdriver).
➤ HEADLESS-режим: запуск без GUI, но его одного часто недостаточно.

Важно: всегда проверяйте robots.txt и Terms of Service сайта. Уважайте нагрузку на чужой сервер (добавляйте задержки time.sleep()).

Приходилось ли вам бороться с анти-бот системами?

Писать парсер для защищенного сайта — это игра в кошки-мышки с системными администраторами этого сайта. Главное — не переходить в DDoS. 🐭🐱


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

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

Скрипты, которые работают с внешними ресурсами (сеть, БД, 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
👍2🤔1
Кэширование для ленивых: ускоряем скрипты в 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
👍2🤔1
Как избежать хаоса: конфигурация скриптов в 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
👍2
Автоматическое тестирование скриптов: оно вам нужно?

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

Скрипты для автоматизации - такое же ПО, как и веб-сервисы. Они могут содержать баги, которые приводят к потере данных или простою систем. Писать тесты для скриптов кажется 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
👍2
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
👍2
Мониторим свое приложение: пишем кастомные метрики для 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
👍1
Кешируем результаты 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
👍4
Пишем простую систему обработки задач с 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
👍2
Пишем свою первую 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
👍21