🐍 Микросервисы на 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