Порт отдал словарь, а не DataFrame — Курс SQL

Чему научишься

смотреть на ответ источника как на словарь со своими правилами, а не как на готовую таблицу
задавать контракт строки: что обязано быть, что нужно привести к типу, а что можно молча проигнорировать
отбраковывать битую запись НА ГРАНИЦЕ, а не обнаруживать её в витрине через три полки
отличать «источник отдал мусор» от «мы уронили загрузку»

Тикет №02. «Падает каждую третью ночь»

11:40. С витриной разобрались. Теперь ты открываешь второй тикет — он висит с прошлой недели, а написал его сам предшественник:

«Загрузчик заказов падает примерно раз в три ночи. В логе всегда одно и то же: KeyError: 'status'. Перезапуск обычно помогает. Разобраться некогда».

«Обычно помогает» здесь не значит «исправляет». При повторе источник отдаёт другой набор строк, и битая запись может просто не попасть в него. Сегодня повезло — загрузка прошла. Завтра та же проблема вернётся.

КВЕРИ: «Перезапуск обычно помогает» — это не диагноз, это ставка. Предшественник делал её раз в три ночи и чаще выигрывал. Тебе ставка не нужна, тебе нужно обещание, которое источник обязан выполнить.

Начнём с самой ошибки. KeyError означает, что код обратился к полю, которого в ответе не было. Загрузчик предполагал: если источник прислал запись, значит, в ней есть всё необходимое. Но источник такого не обещал. Он просто отдал словарь.

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

Сегодня ты строишь первый участок своей ленты — приёмный порт. Его задача не просто «скачать данные», а превратить чужой словарь в нашу строку или честно отказаться.

Сначала посмотрим, что именно приходит на границу.

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

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

from arena_source import fake_api

REQUIRED = ('order_id', 'customer_id', 'city', 'status', 'amount', 'qty', 'created_at')

page = fake_api.get('/orders', {'limit': 120})
rows = page['rows']
print('ответ порта : %s · одна строка: %s · строк: %d · есть ещё: %s'
      % (type(page).__name__, type(rows[0]).__name__, len(rows), page['has_more']))
print()
print('Первая строка, поле за полем:')
for name in sorted(rows[0]):
    print('    %-12s %-4s %r' % (name, type(rows[0][name]).__name__, rows[0][name]))
print()

missing = [r for r in rows if [f for f in REQUIRED if f not in r]]
neg_qty = [r for r in rows if isinstance(r.get('qty'), int) and r['qty'] < 0]
bad_date = [r for r in rows if not str(r.get('created_at', '')).startswith('2184-')]
unknown = [r for r in rows if set(r) - set(REQUIRED) - {'updated_at'}]

print('Из %d строк одной страницы:' % len(rows))
print('    нет обязательного поля : %2d, например %d (нет %s)'
      % (len(missing), missing[0]['order_id'],
         ', '.join(f for f in REQUIRED if f not in missing[0])))
print('    qty отрицательный      : %2d, например %d (qty=%d)'
      % (len(neg_qty), neg_qty[0]['order_id'], neg_qty[0]['qty']))
print('    дата не разбирается    : %2d, например %d (created_at=%r)'
      % (len(bad_date), bad_date[0]['order_id'], bad_date[0]['created_at']))
print('    незнакомое поле        : %2d, например %d (%s)'
      % (len(unknown), unknown[0]['order_id'],
         ', '.join(sorted(set(unknown[0]) - set(REQUIRED) - {'updated_at'}))))

from arena_source import fake_api

REQUIRED = ('order_id', 'customer_id', 'city', 'status', 'amount', 'qty', 'created_at')

page = fake_api.get('/orders', {'limit': 120})
rows = page['rows']
print('ответ порта : %s · одна строка: %s · строк: %d · есть ещё: %s'
      % (type(page).__name__, type(rows[0]).__name__, len(rows), page['has_more']))
print()
print('Первая строка, поле за полем:')
for name in sorted(rows[0]):
    print('    %-12s %-4s %r' % (name, type(rows[0][name]).__name__, rows[0][name]))
print()

missing = [r for r in rows if [f for f in REQUIRED if f not in r]]
neg_qty = [r for r in rows if isinstance(r.get('qty'), int) and r['qty'] < 0]
bad_date = [r for r in rows if not str(r.get('created_at', '')).startswith('2184-')]
unknown = [r for r in rows if set(r) - set(REQUIRED) - {'updated_at'}]

print('Из %d строк одной страницы:' % len(rows))
print('    нет обязательного поля : %2d, например %d (нет %s)'
      % (len(missing), missing[0]['order_id'],
         ', '.join(f for f in REQUIRED if f not in missing[0])))
print('    qty отрицательный      : %2d, например %d (qty=%d)'
      % (len(neg_qty), neg_qty[0]['order_id'], neg_qty[0]['qty']))
print('    дата не разбирается    : %2d, например %d (created_at=%r)'
      % (len(bad_date), bad_date[0]['order_id'], bad_date[0]['created_at']))
print('    незнакомое поле        : %2d, например %d (%s)'
      % (len(unknown), unknown[0]['order_id'],
         ', '.join(sorted(set(unknown[0]) - set(REQUIRED) - {'updated_at'}))))

python · источник

Контракт строки: три решения на каждое поле

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

Например, сумма приехала СТРОКОЙ — '336.68'. Так деньги нередко отдают -API: источник не должен решать за нас, как хранить и округлять значения. Но нашей ленте всё равно нужен конкретный тип.

Пока сумма остаётся строкой, amount * qty даст не выручку, а повтор текста. Поэтому на границе нужно явно превратить внешний формат во внутренний.

Для денег в контракте используем Decimal, а не float. float хранит числа в двоичном формате: 0.1 + 0.2 в нём не равно 0.3, поэтому вычисления с такими значениями могут накапливать погрешность. В stg_orders сумма лежит как numeric(12,2), поэтому контракт порта говорит на том же языке: Decimal('336.68') хранит ровно то, что прислал источник.

С датой та же история. Она тоже приходит строкой, причём в трёх записях из ста двадцати вместо даты стоит слово «вчера». Количество у четырёх записей отрицательное: это сторно, приехавшее отдельной строкой. А поле loyalty_tier источник добавил на прошлой неделе, никого не спросив, — и добавит ещё.

Если реагировать на каждый такой случай уже во время падения, ночная загрузка будет зависеть от случайностей источника. Поэтому для каждого поля решение принимают ЗАРАНЕЕ.

Решение	Что делаем	Пример
обязано быть	нет поля — строка не едет дальше	`order_id`, `status`, `amount`
приводим тип	не приводится — строка не едет дальше	`'336.68'` → `Decimal('336.68')`, `'2184-03-12 01:02:03'` → дата
игнорируем	поля нет в контракте — молча выбрасываем	`loyalty_tier`

Третье решение особенно важно. Новое поле у источника НЕ должно ронять разбор: ночью никто не читает release notes чужой команды. Мы просто не включаем его в свою строку.

А вот пропавшее обязательное поле — уже нарушение контракта. Такую запись нельзя пускать дальше: нужных данных для расчётов всё равно нет.

Сам контракт лучше закрепить в коде, а не оставлять комментарием, который забудут обновить. В Python для этого есть dataclass: перечисленные поля задают форму строки, которую порт обещает следующей полке. Всё, чего нет в dataclass, до stg_orders не доедет — и здесь это не потеря, а намеренное правило границы.

Остаётся решить, что делать с записью, которая контракт не прошла. В этом уроке функция разбора возвращает либо готовую строку, либо None.

None здесь означает не «уронить загрузку», а «эта запись поедет в карантин, а лента продолжит работу». Битые строки на странице из ста двадцати не должны останавливать всю ночь.

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

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

Что здесь настоящее, а что песочница. Python в этом курсе исполняется прямо в браузере, а доступа к сети там нет. Поэтому arena_source.fake_api не ходит в настоящий HTTP.

Но форма взаимодействия с источником сохранена. В этой ячейке важны вызов api.get(path, params) и ответ со списком строк. Признаки has_more и next_cursor, коды ошибок и Retry-After понадобятся в следующих уроках — сейчас их разбирать не нужно. Если заменить fake_api на requests, основные шаги разбора строк останутся теми же.

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

Это важно для отладки: ты можешь сопоставить результат функции с известным набором данных и понять, какое правило контракта сработало.

Практика: напиши код

Напиши приёмный порт: функцию parse(row), которая превращает словарь источника в OrderRow или возвращает None. Задача функции — провести одну запись через контракт и не пропустить дальше то, что ему не соответствует. Контракт, который она обязана держать:

обязательные поля — все семь из REQUIRED. Нет хотя бы одного → None;
amount приходит строкой, в контракте это Decimal (деньги во float не хранят). Не приводится → None;
qty — целое, строго больше нуля. Ноль и минус → None (сторно поедет своей дорогой);
created_at приходит строкой вида 2184-03-12 01:02:03, в контракте это datetime. Не разбирается → None;
незнакомые поля (loyalty_tier, updated_at) разбор не роняют и в OrderRow не попадают.

В заготовке лежит наивный разбор — тот самый, что падал с KeyError. Замени его так, чтобы одна плохая запись превращалась в None, а не останавливала весь порт.

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, Optional

REQUIRED = ('order_id', 'customer_id', 'city', 'status', 'amount', 'qty', 'created_at')


@dataclass
class OrderRow:
    """Контракт строки: то, что порт обещает следующей полке."""
    order_id: int
    customer_id: int
    city: str
    status: str
    amount: Decimal   # деньги — Decimal: в stg_orders это numeric(12,2)
    qty: int
    created_at: datetime


def parse(row: Dict[str, Any]) -> Optional[OrderRow]:
    """Наивный разбор: берёт что дали и ничего не проверяет."""
    return OrderRow(
        order_id=row['order_id'],
        customer_id=row['customer_id'],
        city=row['city'],
        status=row['status'],
        amount=row['amount'],
        qty=row['qty'],
        created_at=row['created_at'],
    )

python · источник

Вопрос с собеседования

Как это спрашивают на собеседовании. «Источник добавил в ответ новое поле и переименовал одно старое. Что должно произойти с вашей загрузкой?»

Здесь важно разделить два события.

Если появилось новое поле, загрузка продолжает работать, а поле игнорируется. Ронять всю ленту из-за того, что источник расширил ответ, нельзя.

Если пропало или было переименовано обязательное поле, строка больше не соответствует контракту. Такие строки отбраковываются и уходят в карантин. Когда из-за массового брака нужно останавливать всю задачу, разберём в главе про контроль качества.

Второй вопрос из той же серии: «где вы отбраковываете брак — на входе или в витрине?» Ответ — на входе.

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

Проверь себя

Источник сохранил все поля контракта и добавил в каждый заказ новое поле delivery_slot. Как обязан вести себя приёмный порт этой ночью?

Проверь себя

В коде разбора написано amount = row['amount'] без приведения типа. Значение 336.68 осталось строкой. Что может произойти, когда код впервые использует его как число?

Закрепление: реши задачи

Решено 0 из 3 · для зачёта достаточно 2

Главное из урока

Источник отдаёт словари, а не готовую таблицу: типы могут быть строковыми, а набор ключей не гарантирован.
Форму строки задаёт dataclass, а правила входа проверяет функция разбора. Для каждого поля заранее выбрано одно из трёх решений: обязано быть, приводим тип или игнорируем.
Незнакомое поле не роняет разбор. Отсутствие обязательного поля роняет СТРОКУ, а не всю ночь.
parse(row) -> OrderRow | None: None означает «в карантин». Решение принимается на границе, пока ещё понятно, что именно прислал источник.
KeyError: 'status' из тикета №02 больше не воспроизводится: наличие поля проверяется до обращения к нему.

Приёмный порт теперь умеет отличать хорошую строку от плохой. Но пока он приносит только первую страницу ответа источника.

Дальше по ленте — : как забрать остальные страницы и не потерять данные между вызовами.