🐍 Надёжные ретраи с backoff и jitter для автоматизации API и задач

👁 В автоматизации частая ошибка делать глупые ретраи: retry 3 раза и всё. В реальности это ломает системы ещё сильнее: если сервис начал падать, ты начинаешь долбить его ещё активнее. Правильный подход — exponential backoff + jitter, чтобы равномерно распределять нагрузку и не создавать «шторма» запросов.

📝 Реализация умных ретраев с backoff и jitter

import time
import random
import requests

def fetch_with_retry(url, retries=5, base_delay=0.5):
    for attempt in range(retries):
        try:
            response = requests.get(url, timeout=2)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            if attempt == retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            jitter = random.uniform(0, delay / 2)
            time.sleep(delay + jitter)

data = fetch_with_retry("https://api.example.com/data")
print(data)

📌 Как это работает:

• Задержка растёт экспоненциально (2^attempt), уменьшая давление на сервис при сбоях.
• jitter добавляет случайность, чтобы избежать синхронных повторов от множества клиентов.
• В случае постоянной ошибки выбрасывается исключение, а не бесконечный цикл.

❗️ В прод-автоматизации ретраи — это не «на всякий случай», а часть архитектуры. Неправильные ретраи могут добить сервис быстрее, чем его изначальная проблема.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Идемпотентные задачи в автоматизации через file locking (защита от дублей)

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

📝 Гарантия единственного запуска через file lock

import fcntl
import time

lock_file = open('/tmp/my_task.lock', 'w')

try:
    fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
    print("Задача запущена")

    time.sleep(10)  # имитация работы

finally:
    fcntl.flock(lock_file, fcntl.LOCK_UN)
    lock_file.close()

📌 Как это работает:

• flock с LOCK_EX | LOCK_NB пытается взять эксклюзивную блокировку без ожидания.
• Если другой процесс уже держит лок — текущий запуск сразу завершится (или можно обработать это как retry).
• Блокировка снимается автоматически при завершении процесса, что защищает от зависших локов.

❗️ В автоматизации идемпотентность — это не «на будущее», а защита от реальных прод-багов. Без блокировок даже простой cron может легко превратиться в источник дублей и гонок.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Разработка приложений на Django с нуля

👁 Курс по Django, который учит создавать полноценные приложения на Python.

Вы разберёте структуру проекта, работу с URL и views, шаблоны и статику, а также освоите ORM, модели и админку. Поймёте, как строится backend и как связывать данные в приложении.

❗️ Подойдёт разработчикам, которые хотят начать работать с Django.

⛓ Изучаем по ссылке

tags: #обучение #django #python #backend

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Атомарные операции с файлами для безопасной автоматизации без битых данных

👁 В автоматизации часто пишут файлы «в лоб»: открыл → записал → закрыл. При сбоях (kill, падение, нехватка места) получаются битые файлы и неконсистентное состояние. В проде это критично: конфиги, кеши, дампы. Правильный подход — атомарная запись через временный файл и rename.

📝 Атомарная запись файла без риска повреждения

import os
import tempfile

def atomic_write(path, data: bytes):
    dir_name = os.path.dirname(path) or "."
    with tempfile.NamedTemporaryFile(dir=dir_name, delete=False) as tmp:
        tmp.write(data)
        tmp.flush()
        os.fsync(tmp.fileno())
        temp_name = tmp.name

    os.replace(temp_name, path)  # атомарная операция

atomic_write("config.json", b'{"key": "value"}')

📌 Как это работает:

• Данные пишутся во временный файл в той же директории, чтобы rename был атомарным.
• fsync гарантирует, что данные реально записаны на диск, а не висят в кеше.
• os.replace заменяет файл мгновенно и безопасно, без промежуточного состояния.

❗️ В автоматизации «записать файл» — это не тривиальная операция. Без атомарности ты рано или поздно словишь битые данные и странные баги, которые очень сложно воспроизводятся.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Ограничение времени жизни кеша в автоматизации через TTLCache без ручной очистки

👁 В автоматизации часто приходится кешировать результаты API, запросов к базе или тяжёлых вычислений. Обычный словарь быстро превращается в проблему: данные устаревают, память растёт, а ручная очистка усложняет код. Намного удобнее использовать TTL-кеш с автоматическим удалением старых значений.

📝 Использование TTLCache для автоматического протухания данных

from cachetools import TTLCache
import time

cache = TTLCache(maxsize=100, ttl=5)

cache["token"] = "abc123"

print(cache.get("token"))

time.sleep(6)

print(cache.get("token"))

📌Как это работает:

• ttl=5 задаёт время жизни значения в секундах, после чего оно автоматически считается устаревшим.
• maxsize ограничивает размер кеша и защищает приложение от бесконтрольного роста памяти.
• cache.get() вернёт None, если значение уже протухло и было удалено из кеша.

❗️ TTL-кеш — простой способ снизить нагрузку на API и ускорить автоматизацию без сложной логики очистки. Особенно полезно для токенов, конфигов и часто повторяющихся запросов.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Ограничение количества одновременно запущенных копий скрипта через межпроцессный семафор

👁 В автоматизации нередко возникают ситуации, когда один и тот же скрипт запускается из cron, CI или планировщика несколько раз одновременно. Это может приводить к дублированию операций, конфликтам при работе с файлами или перегрузке внешних сервисов. Решить проблему можно через межпроцессный семафор.

📝 Защита от параллельного запуска через multiprocessing.Semaphore

from multiprocessing import Semaphore
import time

semaphore = Semaphore(1)

if semaphore.acquire(block=False):
    try:
        print("Скрипт запущен")
        time.sleep(10)
    finally:
        semaphore.release()
else:
    print("Другой экземпляр уже работает")

📌 Как это работает:

• Семафор позволяет ограничить количество одновременно выполняемых экземпляров программы.
• При значении 1 только один процесс сможет получить доступ к защищённому участку кода.
• Остальные запуски сразу получат отказ и смогут завершиться или перейти в режим ожидания.

❗️ Контроль параллельных запусков особенно важен для ETL-процессов, резервного копирования, синхронизации данных и других задач автоматизации, где повторное выполнение может привести к непредсказуемым последствиям.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Создаём веб-приложения на Django

👁 Практический курс по Django для разработки серверной части веб-приложений на Python.

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

❗️ Подойдёт начинающим Python-разработчикам, которые хотят освоить создание веб-приложений на Django.

⛓ Изучаем по ссылке

tags: #обучение #python #django

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Объектно-ориентированное программирование на Python

👁 Курс по ООП в Python, который помогает перейти от процедурного подхода к созданию гибких и масштабируемых программ.

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

❗️ Подойдёт начинающим Python-разработчикам и всем, кто хочет глубже разобраться в принципах объектно-ориентированного программирования.

⛓ Изучаем по ссылке

tags: #обучение #python

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Управляемая параллельность в Python через TaskGroup и контроль отмены задач

👁 Асинхронный код в автоматизации часто ломается не на логике задач, а на их жизненном цикле. В старом подходе с create_task() легко получить ситуацию, где часть корутин продолжает работать после ошибки, а часть исключений просто теряется в фоне. asyncio.TaskGroup решает это через строгую группировку выполнения: либо всё завершается корректно, либо вся группа отменяется целиком.

📝 Управляемый параллелизм с детерминированным поведением

import asyncio

async def job(name, delay):
    await asyncio.sleep(delay)
    if delay > 2:
        raise RuntimeError(f"fail: {name}")
    return f"done {name}"

async def main():
    try:
        async with asyncio.timeout(5):
            async with asyncio.TaskGroup() as tg:
                for i in range(1, 5):
                    tg.create_task(job(f"task-{i}", i))
    except Exception as e:
        print("error:", e)

asyncio.run(main())

📌 Как это работает:
• TaskGroup гарантирует, что все задачи либо завершатся, либо будет поднята ошибка
• при исключении одна задача автоматически останавливает группу
• нет «потерянных» coroutine, как в старом create_task()

❗️ Такой подход делает поведение параллельных задач предсказуемым: если одна задача падает — вся группа корректно отменяется, нет «фантомных» корутин, а тайм-аут гарантирует остановку всей пачки без ручного управления отменой.

tags: #python #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Структуры данных в Python

👁 Курс по базовым и продвинутым структурам данных в Python с упором на практику и решение реальных задач.

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

❗️ Подойдёт Python-разработчикам, которые хотят укрепить понимание алгоритмов и структур данных.

⛓ Изучаем по ссылке

tags: #обучение #python

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Микросервисы на Python

👁 Практический курс по разработке асинхронных микросервисов на Python с использованием современных инструментов.

Разбирает создание сервисов через asyncio и Starlette, взаимодействие компонентов через RabbitMQ и валидацию данных с помощью Pydantic. Показывает тестирование микросервисов с pytest и разработку полноценного чата между пользователем и сервисной архитектурой.

❗️ Подойдёт Python-разработчикам, которые хотят освоить построение распределённых систем и микросервисных приложений.

⛓ Изучаем по ссылке

tags: #обучение #python

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Автоматическое управление повторными попытками через экспоненциальную задержку в Python

👁 При работе с внешними API, очередями и сетевыми сервисами временные ошибки неизбежны: сервер может быть перегружен, соединение может кратковременно упасть или сервис может вернуть ошибку ограничения запросов. Простые повторные вызовы через sleep() создают дополнительную нагрузку и могут только усилить проблему. Более надёжный подход — использовать экспоненциальную задержку с ограничением количества попыток.

📝 Реализация retry-механизма с увеличением интервала ожидания

import time
import random

def retry(func, attempts=5, delay=1):
    for attempt in range(attempts):
        try:
            return func()
        except Exception as error:
            if attempt == attempts - 1:
                raise error

            wait = delay * (2 ** attempt)
            wait += random.uniform(0, 1)

            print(f"retry after {wait:.2f}s")
            time.sleep(wait)

def request():
    if random.random() < 0.7:
        raise RuntimeError("temporary error")
    return "success"

print(retry(request))

📌 Как это работает:

• После каждой ошибки время ожидания увеличивается: 1 → 2 → 4 → 8 секунд, снижая нагрузку на внешний сервис.
• Случайная задержка (jitter) добавляется, чтобы несколько параллельных процессов не повторяли запрос одновременно.
• После превышения количества попыток исключение возвращается выше по стеку для дальнейшей обработки.

❗️ Такой механизм используется в распределённых системах, API-клиентах и фоновых воркерах. Для автоматизации это один из базовых способов сделать обработку временных сбоев более устойчивой и предсказуемой.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 PyCharm для профессиональной разработки на Python

👁 Практический курс по эффективному использованию PyCharm для создания, анализа и поддержки Python-проектов.

В программе разбираются управление проектами разного масштаба, разработка веб-приложений на Django, Flask и других фреймворках, анализ данных, рефакторинг и поиск проблем в коде. Также изучаются Git, GitHub, визуальная отладка, модульное тестирование с pytest и создание собственных Python-пакетов.

❗️ Подойдёт Python-разработчикам, которые хотят ускорить разработку и использовать возможности PyCharm на профессиональном уровне.

⛓ Изучаем по ссылке

tags: #обучение #python #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Автоматический контроль изменений файлов через хеширование содержимого в Python

👁 В автоматизации часто нужно отслеживать изменения конфигураций, резервных копий или наборов данных. Простое сравнение времени изменения файла (mtime) ненадёжно: файл может быть перезаписан без изменения содержимого или изменён с сохранением даты. Более точный подход — вычислять хеш содержимого и сравнивать реальные изменения данных.

📝 Отслеживание изменений файлов через SHA-256

import hashlib
import time
from pathlib import Path


def file_hash(path):
    sha = hashlib.sha256()

    with open(path, "rb") as file:
        for chunk in iter(lambda: file.read(8192), b""):
            sha.update(chunk)

    return sha.hexdigest()


target = Path("/etc/nginx/nginx.conf")
old_hash = file_hash(target)

while True:
    new_hash = file_hash(target)

    if new_hash != old_hash:
        print("file changed:", target)
        old_hash = new_hash

    time.sleep(10)

📌 Как это работает:

• Файл читается небольшими блоками, поэтому метод подходит даже для больших объектов, которые нельзя полностью загрузить в память.
• SHA-256 создаёт уникальный отпечаток содержимого, поэтому изменение данных определяется независимо от времени изменения файла.
• Такой механизм можно использовать как основу для автоматического контроля конфигураций, синхронизации и запуска действий после изменения данных.

❗️ Хеширование файлов — простой, но мощный паттерн для автоматизации. Его используют в системах резервного копирования, CI/CD-процессах и инструментах контроля конфигураций, где важно реагировать именно на изменение содержимого, а не только на изменение метаданных файла.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Разработка пакетов Python для Django

👁 Практический курс по созданию собственных Python-пакетов на базе Django middleware и публикации их в PyPI.

В программе разбирается создание Django-проекта, анализ SQL-запросов приложения, устройство middleware и разработка собственного компонента с базовой функциональностью. В конце курса участники создадут полноценный пакет для Django и научатся публиковать его в Python Package Index.

❗️ Подойдёт Python-разработчикам, которые хотят глубже изучить Django, создавать переиспользуемые компоненты и развивать собственные библиотеки.

⛓ Изучаем по ссылке

tags: #обучение #python #django

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Надёжная автоматизация задач через очередь событий с asyncio.Queue

👁 В автоматизации часто возникает ситуация, когда один процесс создаёт задачи быстрее, чем другой успевает их обработать. Простое создание новых потоков или корутин на каждое событие приводит к неконтролируемому росту нагрузки. asyncio.Queue позволяет разделить генерацию и обработку задач, ограничить параллельность и построить управляемый конвейер.

📝 Создание асинхронного обработчика задач

import asyncio


async def worker(queue):
    while True:
        task = await queue.get()

        try:
            print("processing:", task)
            await asyncio.sleep(1)
        finally:
            queue.task_done()


async def main():
    queue = asyncio.Queue(maxsize=100)

    workers = [
        asyncio.create_task(worker(queue))
        for _ in range(3)
    ]

    for i in range(10):
        await queue.put(f"job-{i}")

    await queue.join()

    for worker_task in workers:
        worker_task.cancel()


asyncio.run(main())

📌 Как это работает:

• asyncio.Queue создаёт буфер между производителями и обработчиками, поэтому скорость поступления задач не ломает обработку.
• maxsize ограничивает количество ожидающих задач и создаёт естественный механизм обратного давления.
• Несколько воркеров позволяют выполнять операции параллельно, сохраняя полный контроль над количеством одновременно работающих задач.

❗️ Такой паттерн используется в фоновых обработчиках, интеграциях с API, обработке файлов и автоматизации инфраструктуры. Вместо хаотичного запуска множества задач получается управляемый конвейер, где нагрузка и выполнение операций остаются предсказуемыми.

tags: #python #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Python от основ до практики

👁 Курс для системного освоения Python — от базового синтаксиса до работы с данными и файловыми форматами.

В программе разбираются синтаксис и структуры данных Python, объектно-ориентированное программирование и модули. Также изучается практическая работа с файлами, CSV, JSON, PDF и ZIP, чтобы полученные знания можно было применять при разработке реальных программ.

❗️ Подойдёт разработчикам, которые хотят систематизировать знания Python и уверенно использовать язык для решения прикладных задач.

⛓ Изучаем по ссылке

tags: #обучение #python

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Автоматический запуск команд с ограничением ресурсов через asyncio и subprocess

👁 В инфраструктурной автоматизации Python часто выступает оркестратором для CLI-инструментов: rsync, ssh, kubectl, terraform, docker. Проблема начинается, когда одна команда зависает или одновременно запускается слишком много процессов. asyncio позволяет управлять десятками внешних команд, ограничивать параллельность и автоматически завершать зависшие операции.

📝 Параллельный запуск команд с таймаутом

import asyncio


async def run(cmd, sem):
    async with sem:
        proc = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        try:
            out, err = await asyncio.wait_for(
                proc.communicate(), timeout=30
            )
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            raise

        return proc.returncode, out, err


async def main():
    sem = asyncio.Semaphore(5)

    jobs = [
        ["ssh", host, "uptime"]
        for host in ["server1", "server2", "server3"]
    ]

    results = await asyncio.gather(
        *(run(job, sem) for job in jobs)
    )

    for code, out, err in results:
        print(code, out.decode().strip())


asyncio.run(main())

📌 Как это работает:

• Semaphore(5) ограничивает количество одновременно работающих процессов, поэтому автоматизация не создаёт неконтролируемую нагрузку на систему или сеть.
• wait_for() устанавливает таймаут для каждой команды. Если внешний процесс зависает, Python принудительно завершает его и освобождает воркер.
• asyncio.gather() позволяет собрать результаты множества операций в одной точке, что удобно для массового выполнения команд на серверах.

❗️ Такой подход превращает Python-скрипт в небольшой оркестратор: можно параллельно выполнять ssh, kubectl, docker или любые другие CLI-команды, сохраняя контроль над количеством процессов и временем выполнения каждой операции.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Python-разработка с нуля

👁 Курс познакомит с основами Python и проведёт от первых программ до создания веб-приложений и работы с базами данных.

Вы освоите синтаксис и структуру Python, библиотеки и инструменты разработки, а также Django для создания веб-приложений. В ходе обучения соберёте портфолио проектов и подготовитесь к работе на фрилансе.

❗️ Подойдёт начинающим разработчикам, которые хотят освоить Python и начать создавать реальные проекты.

⛓ Изучаем по ссылке

tags: #обучение #python

🧭 @recura_tech 🌐 VK 🌐 MAX

🐍 Python может превратить SQLite в надёжную локальную очередь для автоматизации

👁 Если автоматизатор должен переживать перезапуски, падения и временную недоступность внешнего сервиса, хранить задачи только в памяти рискованно.

📝 SQLite позволяет сделать небольшую персистентную очередь без Redis или отдельного брокера

import sqlite3
import time

db = sqlite3.connect("jobs.db", isolation_level=None)

db.execute("""
CREATE TABLE IF NOT EXISTS jobs (
    id INTEGER PRIMARY KEY,
    payload TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    created_at REAL NOT NULL
)
""")

db.execute("""
INSERT INTO jobs (payload, created_at)
VALUES (?, ?)
""", ("deploy server-01", time.time()))

while True:
    with db:
        job = db.execute("""
            SELECT id, payload
            FROM jobs
            WHERE status = 'pending'
            ORDER BY id
            LIMIT 1
        """).fetchone()

        if not job:
            break

        db.execute(
            "UPDATE jobs SET status = 'running' WHERE id = ?",
            (job[0],)
        )

    try:
        print(f"Выполняем: {job[1]}")
        # здесь выполняется реальная задача
        time.sleep(2)

        db.execute(
            "UPDATE jobs SET status = 'done' WHERE id = ?",
            (job[0],)
        )
    except Exception:
        db.execute(
            "UPDATE jobs SET status = 'failed' WHERE id = ?",
            (job[0],)
        )

📌 Как это работает:

• Задача сначала записывается на диск, поэтому перезапуск процесса не уничтожает очередь.
• Состояния pending, running, done и failed позволяют восстановить состояние автоматизации после сбоя.
• SQLite поддерживает транзакции, блокировки и WAL, поэтому для небольших локальных job-очередей отдельный брокер сообщений может быть избыточен.

❗️ Такой подход особенно полезен для CLI-автоматизации, локальных агентов, deploy-утилит и фоновых обработчиков, которым нужна надёжная очередь без дополнительной инфраструктуры.

tags: #python #автоматизация #разработка

🧭 @recura_tech 🌐 VK 🌐 MAX