Как использовать модуль subprocess для запуска внешних команд

Как использовать модуль subprocess для запуска внешних команд

Иногда Python‑кода мало. Нужно дернуть системную утилиту: запустить ping, запустить другой скрипт, конвертировать видео через ffmpeg. Для этого есть модуль subprocess — безопасная и гибкая альтернатива старым os.system.

---

### Простой запуск и получение вывода

Самый удобный вход — функция subprocess.run. Она запускает команду, ждет завершения и возвращает объект с результатом:

import subprocess

result = subprocess.run(
    ["echo", "Hello from subprocess"],
    capture_output=True,
    text=True
)

print("Return code:", result.returncode)
print("Stdout:", result.stdout)
print("Stderr:", result.stderr)

Ключевые моменты:

- Команда передаётся как список: ["echo", "text"], а не одной строкой. Это безопаснее и избавляет от проблем с кавычками.
- capture_output=True собирает stdout и stderr.
- text=True декодирует байты в строки (по умолчанию UTF-8).

---

### Обработка ошибок

Если вы хотите, чтобы Python падал при ошибке команды, используйте check=True:

import subprocess

try:
    subprocess.run(
        ["ls", "/definitely_not_exists"],
        check=True,
        capture_output=True,
        text=True
    )
except subprocess.CalledProcessError as exc:
    print("Command failed!")
    print("Return code:", exc.returncode)
    print("Stderr:", exc.stderr)

CalledProcessError содержит всю полезную информацию о провалившейся команде.

---

### Передача данных на stdin

Иногда внешней программе нужно что‑то «скормить» через стандартный ввод:

import subprocess

text_to_send = "line 1\nline 2\n"

result = subprocess.run(
    ["grep", "line"],
    input=text_to_send,
    text=True,
    capture_output=True
)

print(result.stdout)

Параметр input= передает данные в stdin процесса.

---

### Непосредственное взаимодействие: Popen

Если нужно более тонкое управление (долгоживущий процесс, чтение вывода построчно и т.п.), используйте Popen:

import subprocess

proc = subprocess.Popen(
    ["ping", "-c", "3", "python.org"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

for line in proc.stdout:
    print("PING OUTPUT:", line.strip())

return_code = proc.wait()
print("Return code:", return_code)

Здесь процесс живет «параллельно», а вы можете читать его вывод по мере поступления.

---

### Кратко по безопасности

- Не передавайте в subprocess строки, собранные из непроверенного пользовательского ввода.
- Предпочитайте список аргументов вместо одной строки.
- Не включайте shell=True, если это не абсолютно необходимо. Именно shell=True чаще всего открывает дверь инъекциям команд.

---

subprocess — это мост между вашим Python‑кодом и всем остальным миром командной строки. Освоив его, вы сможете автоматизировать практически любую системную задачу, не покидая Python.

Работа со структурированными логами: использование логгеров с форматами

Python для начинающих: структурированные логи без боли и магии

Обычный print() в продакшене — как дневник на стикерах: пока записей мало, всё понятно, но стоит коду вырасти — и начинается хаос. Здесь на сцену выходит модуль logging и особенно — форматирование логов.

---

### Простейший логгер с форматом

Модуль logging уже в стандартной библиотеке. Начнем с базовой настройки:

import logging

logger = logging.getLogger("app")
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()

formatter = logging.Formatter(
    fmt="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

handler.setFormatter(formatter)
logger.addHandler(handler)

logger.info("Application started")
logger.warning("Low disk space")

Теперь вместо разрозненных сообщений у вас есть чёткая структура: время, уровень, имя логгера, текст.

---

### Логи, которые можно парсить: JSON-формат

Когда логов много, их хочется складывать в системы вроде ELK или Loki. Для этого удобно писать их в JSON.

Минимальный пример без сторонних библиотек:

import logging
import json
import sys
from datetime import datetime

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            "time": datetime.fromtimestamp(record.created).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            log_record["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_record, ensure_ascii=False)

logger = logging.getLogger("json_app")
logger.setLevel(logging.INFO)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)

logger.info("User logged in", extra={"user_id": 42})

Обрати внимание: extra по умолчанию не попадёт в log_record. Чтобы добавлять свои поля, можно расширить JsonFormatter и пройтись по record.__dict__, отфильтровав служебные ключи.

---

### Контекст: больше данных, меньше шума

Ещё одна фишка — LoggerAdapter. Он позволяет приклеить к логам контекст (например, request_id):

import logging

base_logger = logging.getLogger("service")
base_logger.setLevel(logging.INFO)
base_handler = logging.StreamHandler()
base_handler.setFormatter(logging.Formatter("%(levelname)s | %(request_id)s | %(message)s"))
base_logger.addHandler(base_handler)

class RequestAdapter(logging.LoggerAdapter):
    def process(self, msg, kwargs):
        kwargs.setdefault("extra", {})
        kwargs["extra"].setdefault("request_id", self.extra.get("request_id", "-"))
        return msg, kwargs

logger = RequestAdapter(base_logger, {"request_id": "abc-123"})

logger.info("Processing started")

Теперь все логи в рамках запроса будут помечены одним request_id — дебагить станет гораздо проще.

---

Структурированные логи — это не «красивее вывод», а фундамент для нормальной отладки и мониторинга. Настроив форматы однажды, вы перестанете бояться смотреть на логи крупных проектов.

Как написать простой чат-сервер с использованием asyncio

Как написать простой чат-сервер с использованием asyncio

Асинхронность в Python звучит пугающе, но именно она позволяет одному скрипту одновременно обслуживать десятки клиентов. Отличный способ это почувствовать — написать свой мини-чат-сервер на asyncio.

### Идея чата

У нас будет:
- сервер, который принимает подключения;
- список активных клиентов;
- функция, рассылающая сообщения всем подключённым.

Используем высокоуровневый API: asyncio.start_server, StreamReader и StreamWriter.

### Базовый сервер

import asyncio

clients = set()

async def handle_client(reader: asyncio.StreamReader,
                        writer: asyncio.StreamWriter):
    addr = writer.get_extra_info("peername")
    clients.add(writer)
    print(f"Client connected: {addr}")

    try:
        while data := await reader.readline():
            message = data.decode().rstrip()
            if not message:
                continue
            full_msg = f"{addr}: {message}\n"
            await broadcast(full_msg, sender=writer)
    except asyncio.IncompleteReadError:
        pass
    finally:
        print(f"Client disconnected: {addr}")
        clients.remove(writer)
        writer.close()
        await writer.wait_closed()

async def broadcast(message: str, sender: asyncio.StreamWriter | None = None):
    dead = []
    for client in clients:
        if client is sender:
            continue
        try:
            client.write(message.encode())
            await client.drain()
        except ConnectionError:
            dead.add(client)
    for client in dead:
        clients.discard(client)

async def main():
    server = await asyncio.start_server(handle_client, "127.0.0.1", 8888)
    addr = ", ".join(str(sock.getsockname()) for sock in server.sockets)
    print(f"Server started at {addr}")
    async with server:
        await server.serve_forever()

if __name__ == "__main__":
    asyncio.run(main())

### Как это запустить и протестировать

1. Запускаете сервер: python chat_server.py.
2. В другом терминале подключаетесь через telnet или nc:
- telnet 127.0.0.1 8888
или
- nc 127.0.0.1 8888
3. Откройте 2–3 таких клиента и начните печатать сообщения. Всё, что отправляет один клиент, увидят остальные.

### Что важно понять

- async def и await позволяют не блокировать сервер во время ожидания ввода.
- start_server создаёт корутину, которая для каждого клиента запускает handle_client.
- broadcast показывает ключевую идею: один сервер — много клиентов, минимум блокировок.

Дальше можно добавить ники, команды (/quit, /users), простую авторизацию — всё это поверх уже готового асинхронного фундамента.

Изучение шаблонов проектирования на простых примерах: Singleton, Factory

Изучение шаблонов проектирования на простых примерах: Singleton, Factory

В какой‑то момент любой код на Python вырастает до такого размера, что простых функций и классов уже мало. Хочется порядка. Тут в игру выходят шаблоны проектирования. Разберём два базовых — Singleton и Factory — на минимальных, но жизненных примерах.

---

## Singleton: когда нужен только один экземпляр

Singleton гарантирует, что у класса будет всего один объект. Полезно для настроек приложения, логгера, подключения к БД.

### Пример: глобальные настройки приложения

class AppConfig:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialized = False
        return cls._instance

    def __init__(self, debug=False):
        if self._initialized:
            return
        self.debug = debug
        self._initialized = True


cfg1 = AppConfig(debug=True)
cfg2 = AppConfig(debug=False)

print(cfg1 is cfg2)       # True
print(cfg1.debug, cfg2.debug)  # True True

__new__ контролирует создание объекта и хранит единственный экземпляр в _instance.
__init__ запускается при каждом вызове класса, поэтому мы защищаемся флагом _initialized, чтобы не перезаписывать настройки.

---

## Factory: когда нужно создавать разные объекты одинаковым способом

Factory инкапсулирует логику создания объектов. Внешнему коду не важно, какой именно класс используется — он просто просит «сделай мне подходящий объект».

### Пример: парсер данных в разных форматах

class JsonParser:
    def parse(self, data: str):
        return {"type": "json", "raw": data}


class XmlParser:
    def parse(self, data: str):
        return {"type": "xml", "raw": data}


class ParserFactory:
    @staticmethod
    def get_parser(format_name: str):
        format_name = format_name.lower()
        if format_name == "json":
            return JsonParser()
        if format_name == "xml":
            return XmlParser()
        raise ValueError(f"Unknown format: {format_name}")


parser = ParserFactory.get_parser("json")
result = parser.parse('{"id": 1}')
print(result)

Теперь, если появится YamlParser, мы просто добавим его в ParserFactory, не переписывая весь остальной код.

---

Singleton помогает централизовать состояние, а Factory — спрятать логику создания объектов за единым интерфейсом. Оба шаблона отлично тренируют умение думать о структуре программы, а не только о том, «как бы это запустить, чтобы не упало».

Модуль statistics: работа со средним, медианой и дисперсией

Модуль statistics: работа со средним, медианой и дисперсией

Модуль statistics: среднее, медиана и дисперсия без боли в голове

Когда данные растут, как снежный ком, возникает вопрос: «И что со всем этим делать?»
В Python для этого есть модуль statistics — маленький, но очень полезный набор инструментов для анализа чисел.

---

### Среднее: mean

Среднее арифметическое — это классическое "в среднем по больнице".

from statistics import mean

temperatures = [18, 20, 21, 19, 22, 20]
avg_temp = mean(temperatures)
print(avg_temp)  # 20.0

mean() чувствителен к выбросам. Если добавить одно странное значение, картина исказится:

from statistics import mean

salaries = [50_000, 55_000, 52_000, 60_000, 1_000_000]
print(mean(salaries))  # 243000.0 — выглядит уже не так реалистично

---

### Медиана: median и median_low / median_high

Медиана — "середина" отсортированных данных. Она куда устойчивее к выбросам.

from statistics import median

salaries = [50_000, 55_000, 52_000, 60_000, 1_000_000]
print(median(salaries))  # 55_000 — намного ближе к реальности

Для четного количества значений можно управлять поведением:

from statistics import median_low, median_high

values = [1, 2, 100, 101]
print(median_low(values))   # 2
print(median_high(values))  # 100

---

### Дисперсия и стандартное отклонение: насколько данные "разбросаны"

Если среднее говорит "где центр", то дисперсия и стандартное отклонение — "насколько всё вокруг центра разлетается".

from statistics import variance, stdev

scores = [70, 72, 71, 69, 70, 71]
print(variance(scores))  # маленькое значение — все близко к среднему
print(stdev(scores))     # стандартное отклонение

Чем больше дисперсия и стандартное отклонение, тем сильнее отличаются значения внутри набора.

Важно: variance() и stdev() требуют минимум 2 значения. Для всей генеральной совокупности есть pvariance() и pstdev() — используют немного другие формулы:

from statistics import pvariance, pstdev

scores = [70, 72, 71, 69, 70, 71]
print(pvariance(scores))
print(pstdev(scores))

---

### Небольшой практический пример

Оценим стабильность времени отклика сервиса:

from statistics import mean, median, stdev

response_times = [110, 120, 115, 300, 118, 117, 119]

print("mean:", mean(response_times))
print("median:", median(response_times))
print("stdev:", stdev(response_times))

Если среднее сильно больше медианы и стандартное отклонение велико — у вас есть редкие, но тяжелые "подвисания", которые нельзя игнорировать.

---

statistics — идеальный вход в мир анализа данных: минимум кода, максимум информации о ваших числах. Начните с этих функций, и набор цифр перестанет быть просто хаосом.

Как использовать watchdog для отслеживания изменений в файлах

Как использовать watchdog для отслеживания изменений в файлах

Иногда хочется, чтобы скрипт сам реагировал на изменения в файлах: перезапускал сборку, обновлял данные, копировал новый файл в бэкап. Переписывать бесконечные циклы while True с time.sleep — скучно и неэффективно. Для этого есть отличный модуль — watchdog.

---

### Установка

pip install watchdog

watchdog работает кроссплатформенно и умеет «слушать» файловую систему почти как системные утилиты.

---

### Простейший пример: реагируем на изменения в папке

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

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
from pathlib import Path


class SimpleHandler(FileSystemEventHandler):
    def on_modified(self, event):
        print(f"Modified: {event.src_path}")

    def on_created(self, event):
        print(f"Created: {event.src_path}")

    def on_deleted(self, event):
        print(f"Deleted: {event.src_path}")


if __name__ == "__main__":
    path = Path(".").resolve()
    event_handler = SimpleHandler()
    observer = Observer()
    observer.schedule(event_handler, str(path), recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

Теперь, если в папке создать/изменить/удалить файл — скрипт отреагирует.

---

### Фильтрация по типу файлов

Часто нужно отслеживать только, скажем, .py или .txt. Добавим простую проверку:

class PyFilesHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        if not event.src_path.endswith(".py"):
            return
        print(f"Python file changed: {event.src_path}")

Так можно запускать, например, линтер или тесты только при изменении исходников.

---

### Автоматическая реакция: мини‑автосборка

Пример: при изменении файла конфигурации перегенерировать выходной файл.

import subprocess


class BuildHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        if event.src_path.endswith("config.yaml"):
            print("Config changed, rebuilding...")
            subprocess.run(["python", "build.py"])

Теперь любое сохранение config.yaml автоматически запускает build.py.

---

watchdog — мощный инструмент для автоматизации рутинных задач вокруг файловой системы: автосборка проектов, синхронизация папок, логирование изменений. Главное — один раз написать обработчик событий, а дальше файлы будут «говорить» с вашим кодом сами.

Работа с кешированием данных: file-based и memory-based подходы

Работа с кешированием данных: file-based и memory-based подходы
================================================================

Кеширование — это способ «не думать дважды». Если результат уже посчитан, зачем снова тратить время и ресурсы? В Python это особенно актуально при работе с медленными запросами к API, базам данных или тяжёлым вычислениям.

Разберём два простых подхода: кеш в памяти и кеш в файле.

## Memory-based: быстро, но временно

Кеш в памяти живёт только пока работает программа. Зато он очень быстрый.

import time
from functools import lru_cache

@lru_cache(maxsize=128)
def slow_square(n: int) -> int:
    time.sleep(1)  # имитация долгой операции
    return n * n

if __name__ == "__main__":
    start = time.time()
    print(slow_square(10))  # первый вызов — медленно
    print("First call:", time.time() - start)

    start = time.time()
    print(slow_square(10))  # второй вызов — из кеша
    print("Second call:", time.time() - start)

lru_cache:
- автоматически хранит результаты по аргументам;
- ограничивает размер кеша (maxsize);
- идеально подходит для функций без побочных эффектов (чистые функции).

Минус: как только программа завершилась — кеш исчез.

## File-based: медленнее, но переживает перезапуск

Кеш в файле сохраняет результаты между запусками. Удобно для скриптов, которые периодически запускаются по расписанию.

Самый простой способ — использовать pickle или json. Пример с json:

import json
import os
import time
from typing import Any, Dict

CACHE_FILE = "cache.json"

def load_cache() -> Dict[str, Any]:
    if not os.path.exists(CACHE_FILE):
        return {}
    with open(CACHE_FILE, "r", encoding="utf-8") as f:
        return json.load(f)

def save_cache(cache: Dict[str, Any]) -> None:
    with open(CACHE_FILE, "w", encoding="utf-8") as f:
        json.dump(cache, f)

def get_data(param: str) -> str:
    time.sleep(1)  # имитация тяжёлого запроса
    return f"result_for_{param}"

def cached_get_data(param: str) -> str:
    cache = load_cache()
    if param in cache:
        return cache[param]

    result = get_data(param)
    cache[param] = result
    save_cache(cache)
    return result

if __name__ == "__main__":
    print(cached_get_data("foo"))  # первый раз — долго
    print(cached_get_data("foo"))  # второй раз — сразу

Плюсы:
- данные сохраняются между запусками;
- можно легко чистить/анализировать кеш-файл.

Минусы:
- медленнее RAM;
- нужно следить за размером файла и «протуханием» данных.

## Что выбрать?

- Нужна скорость внутри одного процесса — берите lru_cache.
- Важно экономить время между запусками — храните кеш в файле.
- В реальных проектах их часто комбинируют: быстрый memory-кеш поверх более «упорного» file-кеша.

Кеширование — один из тех инструментов, которые мгновенно делают ваш код ощущаемо «умнее» и быстрее, если применять его осознанно.

Как исключить дубли в данных с помощью множеств

Как исключить дубли в данных с помощью множеств

Дубли в данных — классическая боль новичка: выгрузили список email’ов, пользователей или айдишников, а там половина повторяется. Перебирать руками? Нет. В Python есть более элегантный инструмент — множества (set).

### Что такое множество?

set — это неупорядоченная коллекция уникальных элементов. Любой дубль при добавлении просто игнорируется.

emails = ["a@example.com", "b@example.com", "a@example.com", "c@example.com"]
unique_emails = set(emails)

print(unique_emails)
# {'a@example.com', 'b@example.com', 'c@example.com'}

Бонус: операции над множествами быстрые. Если нужно часто проверять, есть ли элемент в коллекции, set обычно быстрее списка.

---

### Удаляем дубли, но сохраняем порядок

У set нет порядка, а иногда он важен (например, порядок регистрации пользователей). Тогда можно комбинировать список и множество:

users = ["alice", "bob", "alice", "dave", "bob", "eve"]

seen = set()
unique_users = []

for user in users:
    if user not in seen:
        seen.add(user)
        unique_users.append(user)

print(unique_users)
# ['alice', 'bob', 'dave', 'eve']

seen отвечает только за проверку дублей, а итоговый порядок хранится в unique_users.

---

### Убираем дубли в списке словарей

Сложнее случай: у нас есть список словарей, и мы хотим убира́ть дубли по какому-то полю, например id.

records = [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
    {"id": 1, "name": "Alice Smith"},
    {"id": 3, "name": "Eve"},
]

seen_ids = set()
clean_records = []

for record in records:
    record_id = record["id"]
    if record_id not in seen_ids:
        seen_ids.add(record_id)
        clean_records.append(record)

print(clean_records)
# [{'id': 1, 'name': 'Alice'},
#  {'id': 2, 'name': 'Bob'},
#  {'id': 3, 'name': 'Eve'}]

Мы не пытаемся сделать множество из словарей (они неизменяемыми быть не могут), мы используем множество только для хранения уже встреченных id.

---

### Быстрая проверка пересечений

Множества полезны не только для удаления дублей, но и для анализа данных: пересечения, разности, объединения.

old_users = {"alice", "bob", "carol"}
new_users = {"bob", "dave", "eve"}

returned_users = old_users & new_users      # пересечение
all_users = old_users | new_users           # объединение
lost_users = old_users - new_users          # кто не вернулся

print(returned_users)  # {'bob'}
print(all_users)       # {'alice', 'bob', 'carol', 'dave', 'eve'}
print(lost_users)      # {'alice', 'carol'}

---

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

Работа с псевдослучайным выбором с помощью random.choices

Python для начинающих: магия random.choices — взвешенный случайный выбор

Когда мы слышим «случайность в Python», чаще всего вспоминаем random.choice. Он берет один случайный элемент из последовательности — и на этом всё.
Но как только появляется слово «вероятность», в игру вступает его более продвинутый родственник — random.choices.

## Базовый пример: не просто выбрать, а выбрать с шансами

Функция random.choices умеет:
- выбирать несколько элементов сразу;
- учитывать «вес» каждого элемента (то есть вероятность).

import random

fruits = ["apple", "banana", "orange"]
weights = [1, 3, 6]  # вероятности: apple < banana < orange

result = random.choices(fruits, weights=weights, k=10)
print(result)

Здесь k=10 — сколько элементов выбрать.
orange будет выпадать чаще всего, потому что его вес больше.

## Чем choice отличается от choices

- random.choice(seq) → один элемент, все равновероятны.
- random.choices(seq, k=n) → список из n элементов, с возможностью задавать веса.

items = ["A", "B", "C"]

print(random.choice(items))           # один элемент
print(random.choices(items, k=5))     # список из 5 элементов

По умолчанию choices тоже считает все элементы равновероятными, пока не указаны weights или cum_weights.

## Веса и кумулятивные веса

weights — обычные веса (не обязаны быть в сумме 1):

cards = ["common", "rare", "epic", "legendary"]
weights = [70, 20, 9, 1]  # проценты можно записать и так, пропорционально

pulled = random.choices(cards, weights=weights, k=20)
print(pulled)

cum_weights — кумулятивные (накопленные) значения. Например, [70, 90, 99, 100] — та же таблица вероятностей, но в формате «до этой границы»:

cards = ["common", "rare", "epic", "legendary"]
cum_weights = [70, 90, 99, 100]

pulled = random.choices(cards, cum_weights=cum_weights, k=20)
print(pulled)

weights и cum_weights вместе указывать нельзя — выбери что-то одно.

## Выбор с возвращением и без

random.choices всегда выбирает с возвращением — один и тот же элемент может встретиться несколько раз. Это удобно для моделирования повторяющихся событий: бросков, попыток, лут-боксов.

Если нужен выбор без повторов, смотри на:
- random.sample(population, k) — без повторов;
- или на random.shuffle и срезы.

import random

numbers = list(range(10))

print(random.sample(numbers, 3))  # 3 уникальных числа

## Минииcследование: проверяем, что веса работают

import random
from collections import Counter

choices = ["A", "B", "C"]
weights = [1, 2, 7]

result = random.choices(choices, weights=weights, k=10000)
counts = Counter(result)

for item in choices:
    print(item, counts[item] / 10000)

Запустив этот код, ты увидишь, что частоты примерно соответствуют пропорциям 1:2:7.

random.choices — отличный инструмент, когда нужно не просто «случайно», а «случайно с контролем»: симуляции, простые игровые механики, генерация тестовых данных — везде, где вероятность чего-то должна быть не только ощущаемой, но и задаваемой.

Создание API-клиентов: практика на примере GitHub API