Скрипт как сервис: структура, argparse, logging, коды завершения

Скрипт загрузки живёт не в ноутбуке, а в расписании: его запускает Airflow или cron, читает его вывод человек в три часа ночи, и по его коду завершения оркестратор решает, упала задача или нет. Разница между «скриптом» и «утилитой» не в объёме кода, а в четырёх договорённостях: как запускать, как передать параметры, как читать вывод, как понять результат.

Точка входа

def main(argv=None) -> int:
    args = parse_args(argv)
    configure_logging(args.verbose)
    try:
        run(args)
    except Exception:
        log.exception("загрузка упала")
        return 1
    return 0
 
if __name__ == "__main__":
    sys.exit(main())

if __name__ == "__main__" отделяет запуск файла от его импорта: при python load.py блок выполнится, при from load import run нет. Без этой проверки тесты, которые импортируют модуль, запустят загрузку.

main принимает argv параметром и возвращает код, а не вызывает sys.exit внутри. Так его можно вызвать из теста с нужными аргументами и проверить код возврата, не перехватывая SystemExit. Вся логика живёт в run, main только обвязка: разобрать аргументы, настроить логи, поймать ошибку, вернуть код.

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

Дата, за которую грузим, путь к файлу, флаг «только проверить» это аргументы, а не константы в коде и не input(). Стандартный модуль argparse:

import argparse
 
def parse_args(argv=None):
    parser = argparse.ArgumentParser(description="Загрузка заказов в витрину")
    parser.add_argument("--date", required=True, help="день загрузки, YYYY-MM-DD")
    parser.add_argument("--batch-size", type=int, default=1000)
    parser.add_argument("--dry-run", action="store_true", help="не писать в базу")
    return parser.parse_args(argv)

parse_args(argv) с None читает sys.argv, со списком читает его: это и есть точка для тестов. type=int приводит и проверяет значение, required=True даёт понятную ошибку, --help генерируется сам. На неправильных аргументах argparse печатает подсказку и завершает процесс с кодом 2.

В Airflow аргументы приходят из шаблонов: --date {{ ds }}. Поэтому дата в утилите всегда параметр, а не date.today(): перезапуск за прошлый день должен грузить прошлый день.

Логирование вместо print

print пишет в stdout без времени, уровня и имени модуля, и его нельзя отключить или перенаправить. logging даёт всё это:

import logging
 
log = logging.getLogger("load_orders")
 
def configure_logging(verbose=False):
    logging.basicConfig(
        level=logging.DEBUG if verbose else logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
 
log.info("загружено %d строк за %.1f с", rows, took)
log.warning("пропущено %d плохих строк", bad)
log.exception("не удалось записать батч")     # внутри except: добавит traceback

Уровни это фильтр: DEBUG для отладки, INFO для хода работы, WARNING для того, что не сломало загрузку, но требует внимания, ERROR для падений. В проде обычно INFO, и DEBUG включается флагом. Аргументы передаются через %s, а не f-строкой: строка форматируется только если уровень включён.

basicConfig настраивается один раз в main, библиотечные модули только получают логгер по имени и пишут. Логгер по имени модуля (logging.getLogger(__name__)) даёт в логе понятно, откуда запись. Ключевые цифры (сколько прочитано, сколько записано, сколько отклонено) логируются всегда: по ним потом ищут, когда что-то пошло не так.

Секреты в лог не попадают. Строка подключения с паролем, токен API, персональные данные из строки, которая не разобралась, всё это в логе навсегда.

Код завершения

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

Частичная загрузка с отклонёнными строками это тоже решение о коде: если отклонено больше порога, код 1, даже если всё, что можно, записано. Договорённость про порог должна быть в аргументах или конфиге, а не в голове автора.

Конфигурация

Параметры запуска (дата, файл) идут аргументами. Параметры окружения (адрес базы, токен) идут переменными окружения: os.environ["DWH_DSN"] с понятной ошибкой, если переменной нет. В Airflow это Connections и Variables, в Docker env, в cron файл, который source перед запуском. В коде и в репозитории значений нет.

Модуль структурно делится так: parse_args, configure_logging, чистые функции трансформации, run который их связывает, main. Чистые функции тестируются без базы и файлов, run с подменёнными зависимостями, main одним тестом на код возврата. Об этом следующий урок.

Как отвечать на собеседовании

Зачем if __name__ == "__main__"? Чтобы код запуска не выполнялся при импорте модуля: тесты и другие модули импортируют функции, а запуск происходит только при вызове файла напрямую.

Почему logging, а не print? Уровни, время, имя модуля, отключение и перенаправление без правки кода, exception с traceback. Настройка один раз в точке входа, модули только пишут.

Как оркестратор узнаёт, что задача упала? По коду завершения процесса: 0 успех, иначе ошибка. Поэтому исключение на верхнем уровне логируется и превращается в ненулевой код, а не глотается.

Откуда скрипт берёт дату и адрес базы? Дата и прочие параметры запуска из аргументов командной строки, чтобы перезапуск за прошлый день был воспроизводим. Адреса и секреты из переменных окружения, не из кода.

Практика

import argparse