lab00 — Python для эксплуатации¶
Стартовая лаборатория для тех, кто идёт в SRE/DevOps. Структуры данных — это середина пути; человек из Ops спотыкается раньше: на файлах, аргументах командной строки и исключениях. Эта лаба закрывает разрыв — на ней «массив без len» из lab04 перестаёт выглядеть абстрактным упражнением и становится подступом к рабочему скрипту.
Четыре темы, каждая — кирпич любого эксплуатационного скрипта:
- Файлы через контекстные менеджеры (
with). - Аргументы командной строки на
argparseи коды возврата процесса. - Исключения как часть контракта функции (собственный класс
ConfigError). - Генераторы и итераторы — ленивая обработка, зацепка за самописный
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), а интерактив в тестах не участвует.