Перейти к содержанию

lab00 — Python для эксплуатации

Стартовая лаборатория для тех, кто идёт в SRE/DevOps. Структуры данных — это середина пути; человек из Ops спотыкается раньше: на файлах, аргументах командной строки и исключениях. Эта лаба закрывает разрыв — на ней «массив без len» из lab04 перестаёт выглядеть абстрактным упражнением и становится подступом к рабочему скрипту.

Четыре темы, каждая — кирпич любого эксплуатационного скрипта:

  1. Файлы через контекстные менеджеры (with).
  2. Аргументы командной строки на argparse и коды возврата процесса.
  3. Исключения как часть контракта функции (собственный класс ConfigError).
  4. Генераторы и итераторы — ленивая обработка, зацепка за самописный CustomRange.

Файлы и контекстные менеджеры

Контекстный менеджер (with) гарантирует, что файл закроется и буфер сбросится на диск даже если внутри блока вылетит исключение. Ручной close() легко забыть или потерять при ошибке — with снимает эту заботу.

def write_lines(path, lines) -> int:
    count = 0
    with open(path, "w", encoding="utf-8") as f:   # файл закроется сам
        for line in lines:
            f.write(line + "\n")
            count += 1
    return count


def read_lines(path) -> list[str]:
    result = []
    with open(path, encoding="utf-8") as f:
        for line in f:
            result.append(line.rstrip("\n"))       # убираем только перевод строки
    return result

read_lines читает файл целиком в память — это нормально для небольших конфигов, но опасно для многогигабайтных логов. Для них есть ленивый вариант (ниже).

Генераторы и итераторы

Генератор — функция с yield: её тело не выполняется при вызове, только при итерации. Значения отдаются по одному, поэтому весь набор данных не висит в памяти.

def iter_lines(path):
    with open(path, encoding="utf-8") as f:
        for line in f:
            yield line.rstrip("\n")        # отдаём строку и «замораживаемся» до next()

iter_lines читает файл построчно и лениво: в памяти живёт одна строка за раз. На него можно навесить фильтрующий генератор — получится «ленивый grep», который не материализует промежуточный список:

def filter_lines(lines, needle):
    for line in lines:
        if needle in line:
            yield line

# потоковая цепочка: читаем и фильтруем, не загружая файл целиком
for line in filter_lines(iter_lines("huge.log"), "ERROR"):
    print(line)

Зацепка за CustomRange из common: самописный ленивый range ведёт себя в генераторе так же, как встроенный — ничего не строится, пока не началась итерация.

def even_squares(stop):
    for n in custom_range(stop):           # ленивый range из labs.common
        if n % 2 == 0:
            yield n * n
Подход Память Когда
read_lines (список) O(n) — весь файл мелкие файлы, нужен случайный доступ / длина
iter_lines (генератор) O(1) — одна строка большие файлы, потоковая обработка

Исключения как часть контракта

Хорошая функция сообщает об ошибке внятно, а не падает с загадочным трейсбеком в середине чужого кода. Для этого заводим своё исключение — тогда вызывающий ловит именно его и понимает, о чём речь.

class ConfigError(Exception):
    def __init__(self, message, line_no=None, line=None):
        self.line_no = line_no
        self.line = line
        if line_no is not None:
            message = f"строка {line_no}: {message}"
        super().__init__(message)

Своё ConfigError (а не голый ValueError) делает контракт явным: «эта функция при кривом конфиге бросит ConfigError, и в нём будет номер строки». Разбор key=value явно перечисляет, что считается ошибкой:

def parse_config_line(line, line_no=None):
    stripped = line.strip()
    if stripped == "" or stripped.startswith("#"):
        return None                                  # пропускаем пустые и комментарии
    if "=" not in stripped:
        raise ConfigError("ожидается формат key=value", line_no, line)
    key, _, value = stripped.partition("=")
    key, value = key.strip(), value.strip()
    if key == "":
        raise ConfigError("пустой ключ", line_no, line)
    return key, value

Контракт parse_config_line в трёх строках:

Вход Результат
пустая строка / # комментарий None (строку пропускаем)
корректная key = value ("key", "value")
нет = или пустой ключ raise ConfigError

load_config читает файл через with и отдаёт текст в parse_config — ошибка укажет на конкретную строку файла.

Аргументы командной строки и коды возврата

argparse превращает «сырые» аргументы в объект настроек, сам обрабатывает --help и неверный ввод. Ключ к тестируемости — разделить разбор и работу: parse_args(argv) принимает список строк (а не лезет в sys.argv), а run(args) делает дело и возвращает код возврата процесса.

def parse_args(argv=None):
    parser = argparse.ArgumentParser(prog="lab00")
    parser.add_argument("path")
    parser.add_argument("-c", "--count", action="store_true")
    parser.add_argument("-g", "--grep", metavar="ПОДСТРОКА")
    return parser.parse_args(argv)        # argv явный — в тесте передаём список

run возвращает int — это и есть код, который уходит в оболочку через sys.exit(...). В Unix 0 — успех, ненулевое — конкретная ошибка:

def run(args) -> int:
    if not args.count and args.grep is None:
        print("Нужно указать --count или --grep ПОДСТРОКА.")
        return 2                          # 2 — не сказано, что делать
    path = Path(args.path)
    try:
        ...                               # потоковая обработка поверх iter_lines
    except FileNotFoundError:
        print(f"Файл не найден: {path}")
        return 1                          # 1 — файла нет
    return 0                              # 0 — успех
Код возврата Значение
0 успех
1 файл не найден / не читается
2 конфликт аргументов (не задано действие)

Почему это важно: код возврата — то, как скрипт общается с оболочкой и CI. command && echo ok, set -e, healthcheck в контейнере, шаг пайплайна — все смотрят именно на код возврата, а не на текст в stdout. Утилита, которая всегда возвращает 0, ломает автоматизацию молча.

Разделение parse_args / run делает обе части чистыми: аргументы проверяются без файловой системы, а run тестируется на временных файлах без подмены sys.argv.

Где это в проде

Эта лаба — скелет любого эксплуатационного скрипта; ровно из этих кусков они и собираются:

  • Парсер конфига. load_config + ConfigError — это мини-версия того, как утилиты читают app.conf / .env / ini и падают с понятным сообщением «строка 7: ожидается key=value» вместо трейсбека. Так ведут себя nginx -t, sshd -t, валидаторы Ansible.
  • Обёртка над subprocess. parse_args + run с кодами возврата — каркас CLI-обёртки (бэкап, деплой, миграция): разобрали аргументы, сделали работу, вернули код возврата, чтобы set -e и шаг пайплайна узнали об ошибке.
  • Healthcheck. Ленивый iter_lines + filter_lines (потоковый «grep по логу») и код возврата 0/1 — основа проверки живости: пройтись по свежим строкам лога, найти ERROR, вернуть ненулевой код, если что-то не так. Это то, на что смотрят HEALTHCHECK в Docker, readinessProbe в k8s и systemd-юниты.

Меню

menu создаёт временную папку (tempfile.TemporaryDirectory), чтобы демонстрации не засоряли репозиторий, и в цикле читает выбор:

Пункт Действие
1 Файлы: запись, чтение целиком и ленивое потоковое чтение с фильтром
2 Конфиг: разбор key=value и перехват ConfigError на битой строке
3 Генераторы: even_squares поверх CustomRange
4 CLI: parse_args + run и коды возврата (0, 1)
0 Выход

Меню — только демонстрация: вся логика живёт в чистых функциях, которые покрыты тестами (файлы — через tmp_path, argparse — через parse_args/run, исключения — через pytest.raises), а интерактив в тестах не участвует.