Зачем учить Python, если агент уже пишет пайплайны?

Допустим, агент написала загрузку заказов. Есть функции, аннотации типов, логирование и даже тесты. На небольшом файле всё работает.

Можно ли теперь не разбираться в Python?

Я бы сначала задал три вопроса. Что останется после падения посреди записи? Что произойдёт при повторном запуске? И зачем эта строчка складывает всю выгрузку в память?

Точный синтаксис можно подсмотреть. А вот заметить, что здесь вообще есть проблема, без понимания кода сложнее.

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

Разберём несколько примеров. Автор кода здесь неважен: человек и модель проходят одно ревью.

1. Файл небольшой. Пока

Нужно посчитать общую сумму заказов из CSV. По условиям примера поле amount_kopecks содержит целое число копеек.

import csv


with open("orders.csv", encoding="utf-8", newline="") as src:
    rows = list(csv.DictReader(src))

total = sum(int(row["amount_kopecks"]) for row in rows)

Код делает то, что написано. В том числе создаёт список со всеми строками файла. DictReader умеет выдавать записи по одной, но вызов list() собирает их в памяти. Для маленькой выгрузки это нормально. Для большой — уже решение, которое нужно обосновать. docs.python.org/3/library/csv

Для подсчёта суммы хранить все заказы не требуется:

with open("orders.csv", encoding="utf-8", newline="") as src:
    total = sum(
        int(row["amount_kopecks"])
        for row in csv.DictReader(src)
    )

Теперь читаем запись, прибавляем сумму и идём дальше. Коллекцию всех заказов не создаём. При этом вычисление остаётся внутри with: файл нужен до завершения чтения.

Это не универсальный запрет на списки. Иногда данные действительно нужно сохранить для повторного обхода или другой обработки.

Вопрос на ревью звучит не «почему здесь list()?», а «зачем здесь одновременно нужны все строки?»

Если ответ — «модель так написала», с оперативной памятью ещё не договорились.

2. Ошибка в логе, успех в Airflow

Пусть load_orders() — функция загрузки. Внутри задачи её вызов обернули так:

import logging


logger = logging.getLogger(__name__)

try:
    load_orders()
except Exception:
    logger.exception("Не удалось загрузить заказы")

В обработчике исключение записывается в лог вместе со стеком вызовов. Но дальше оно не передаётся: после except выполнение продолжается. Если это обычная Python-задача Airflow и других ошибок нет, такой обработчик позволяет ей завершиться успешно. docs.python.org/3/library/logging

Загрузка сломалась. Мониторинг не расстроили. Все молодцы, кроме данных.

Когда ошибка должна остановить задачу, её нужно пробросить:

try:
    load_orders()
except Exception:
    logger.exception("Не удалось загрузить заказы")
    raise

raise без аргументов повторно выбрасывает текущее исключение. А если обработчик не добавляет полезного контекста и ничего не восстанавливает, можно вообще убрать эту обёртку и дать ошибке выйти наружу. docs.python.org/3/tutorial/errors

Разумеется, не любое исключение требует падения всей загрузки. Например, по условиям задачи одну некорректную запись можно отправить в отдельное хранилище ошибок и продолжить.

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

3. Повторили загрузку — увеличили продажи

Представим другой сценарий. Загрузка успела записать часть заказов, затем соединение оборвалось. Задачу запустили повторно, и она снова добавила те же записи в таблицу без ограничения уникальности.

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

В рекомендациях Airflow отдельно сказано: задачи должны корректно переживать повторный запуск. Среди предложенных способов — использовать UPSERT вместо безусловного добавления строк и работать с конкретным интервалом данных, а не с постоянно меняющимся «самым свежим». airflow.apache.org/docs/apache-airflow/stable/best-practices

Для небольшого примера на PostgreSQL создадим таблицу:

CREATE TABLE orders (
    order_id bigint PRIMARY KEY,
    amount_kopecks bigint NOT NULL
);

А затем будем вставлять заказ или обновлять его сумму:

INSERT INTO orders (order_id, amount_kopecks)
VALUES (42, 19900)
ON CONFLICT (order_id) DO UPDATE
SET amount_kopecks = EXCLUDED.amount_kopecks;

При конфликте по order_id PostgreSQL обновит существующую строку. Повтор того же INSERT сохранит для заказа ту же сумму, а не создаст второй заказ. postgresql.org/docs/current/sql-insert

Но само наличие ON CONFLICT ещё ничего не доказывает. Заменим присваивание на прибавление:

SET amount_kopecks =
    orders.amount_kopecks + EXCLUDED.amount_kopecks;

Теперь каждый повтор снова увеличивает сумму. Синтаксис правильный. Дублирующей строки нет. Результат неправильный.

Это уже можно проверить прямо по выражению: мы не устанавливаем значение, а меняем его относительно предыдущего состояния.

У первого варианта тоже есть границы. Например, старое событие может перезаписать более свежую сумму. А триггеры могут создавать дополнительные записи при каждом обновлении. Эти случаи требуют отдельных правил и тестов.

Да, пример получился на SQL. Именно поэтому знание Python не сводится к знанию Python: нужно понимать и операции, которые он запускает.

4. Новая выгрузка не дописалась. Старая уже исчезла

Пишем новую выгрузку поверх старого файла:

with open("orders.csv", "w", encoding="utf-8") as dst:
    dst.writelines(source_lines)

Режим "w" обнуляет существующий файл при открытии. Если после этого источник выдаст часть данных и упадёт, прежней выгрузки уже не будет. Вместо неё останется неполный результат. docs.python.org/3/library/functions

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

Сохраним функцию в publisher.py:

import os
from collections.abc import Iterable
from pathlib import Path
from tempfile import TemporaryDirectory


def publish_lines(lines: Iterable[str], target: Path) -> None:
    target.parent.mkdir(parents=True, exist_ok=True)

    with TemporaryDirectory(dir=target.parent) as tmp_dir:
        draft = Path(tmp_dir) / target.name

        with draft.open("w", encoding="utf-8", newline="") as dst:
            dst.writelines(lines)

        os.replace(draft, target)

Функция принимает строки с уже расставленными переводами строк. До завершения записи целевой файл не трогаем. Временный каталог создаём рядом с ним, чтобы замена происходила в пределах одной файловой системы.

На POSIX-системах успешный os.replace() выполняет атомарную замену: читатель, открывающий целевой путь, получает старую либо новую версию, а не промежуточный недописанный файл. TemporaryDirectory убирает временный каталог при выходе из блока, в том числе при обычном исключении. docs.python.org/3/library/os

Здесь важно не дорисовать гарантии, которых в примере нет. Мы рассматриваем один локальный файл и одного писателя. Функция не решает конкуренцию нескольких загрузок и не реализует отдельный протокол сохранности при отключении питания. Для объектного хранилища схему публикации нужно проектировать отдельно.

Есть ещё одна неприятная деталь: источник может вернуть 40 строк вместо 100 и не выбросить исключение.

Эта функция сама об этом не узнает. Проверка полноты должна происходить до os.replace() — например, сверка количества записей с данными источника. Успешно записанный файл и полная выгрузка — разные условия.

5. Проверяем не обещание функции, а последствия сбоя

Для функции выше полезен конкретный тест: источник оборвался после первой строки, но старая выгрузка сохранилась, а временный каталог убрался.

В test_publisher.py:

from collections.abc import Iterator
from pathlib import Path

import pytest

from publisher import publish_lines


def test_failed_publish_keeps_old_file(tmp_path: Path) -> None:
    target = tmp_path / "orders.csv"
    target.write_text("old data\n", encoding="utf-8")

    def broken_source() -> Iterator[str]:
        yield "new data\n"
        raise RuntimeError("Источник оборвал соединение")

    with pytest.raises(RuntimeError, match="оборвал соединение"):
        publish_lines(broken_source(), target)

    assert target.read_text(encoding="utf-8") == "old data\n"
    assert sorted(tmp_path.iterdir()) == [target]

Здесь tmp_path предоставляет отдельный временный каталог для теста, а pytest.raises() проверяет ожидаемое исключение. Запуск при установленном pytest: python -m pytest -q. docs.pytest.org/en/stable/how-to/tmp_path

Главное в тесте — не вызов pytest.raises(). Мы проверяем состояние после ошибки: что осталось на диске и что увидит следующий потребитель.

Такой тест не доказывает надёжность всей загрузки. Например, он ничего не говорит про неожиданно пустую выгрузку или два параллельных запуска. Зато отвечает на конкретный вопрос, ради которого мы меняли реализацию.

Нейросети вполне можно поручить написать этот тест. Но сначала нужно сформулировать ожидаемое поведение. Иначе получится замкнутый кружок: модель написала реализацию, под неё написала проверку и подтвердила, что они согласны друг с другом.

Сколько Python нужно знать

Для начала я бы ориентировался на три группы навыков:

• Чтение кода: функции, коллекции, изменяемые объекты, итераторы, исключения и with. Нужно понимать, когда выполняется операция, что хранится в памяти и куда уходит ошибка.

• Работа с внешними системами: файлы, форматы данных, запросы к API, тайм-ауты, соединения с БД и транзакции. Здесь учим не только Python, но и поведение используемых инструментов.

• Проверка результата: тесты, отладчик, чтение стека ошибки, контроль входных и выходных данных. Особенно после частичного сбоя и повторного запуска.

Это не требование сначала выучить весь язык и только потом получить разрешение на первый CSV. Метаклассы подождут. Потерянная выгрузка — обычно нет.

Хорошее учебное упражнение — небольшая загрузка из CSV в PostgreSQL. Сначала обычные данные. Потом дубликат заказа, некорректная сумма, обрыв посреди загрузки и повторный запуск.

Перед каждым экспериментом запиши, какой результат считаешь правильным. Должна ли загрузка остановиться? Можно ли пропустить запись? Что должно остаться в таблице?

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

Умение воспроизвести вчерашний ответ модели без подсказки — сомнительный экзамен. Умение объяснить, почему повторный запуск не испортит данные, — уже полезный.

Что в итоге?

Я бы не ставил задачу соревноваться с нейросетью в скорости набора кода. Лучше научиться проверять то, что она написала: где закончится память, какое исключение потеряется, что останется после сбоя.

Сгенерированную загрузку всё равно придётся встроить в конкретную систему с конкретными данными и ограничениями.

И хорошо бы разобраться с этим до того, как бизнес спросит, почему вчерашняя выручка выросла после перезапуска DAG.

Ссылки

• Повод для статьи — материал Joseph Machado Do I Need to Learn Python for Data Engineering, Now That We Have AI?

• tg