Python кодогенерация на благо ETL

В процессе разработки весьма часто встаёт задача преобразования данных, будь то данные от внешнего источника на пути в базу или данные из базы на пути в отчеты и т.п. Если описывать все необходимые преобразования императивно, то можно довольно скоро загрустить. Можно постараться и сделать всё декларативно, скажем, в виде некоторых dict-ов, в которых задать правила (функции?) по работе с каждым отдельным полем. Но уже на этом этапе появляется несколько проблем:

• даже если красиво уместить описание необходимых агрегаций в вышеупомянутый dict, то встроенный itertools.groupby требует предварительно отсортированных данных (порой проще считать, что его нет)

• при обработке, скажем, табличного отчета, на каждой строке будет запускаться дополнительный цикл обхода dict-а с правилами по работе с каждым полем, да еще и дополнительные вызовы лишних ф-й на каждом поле — это замедляет процедуру

• в python отсутствует встроенный функционал join (пока не завернете во что-нибудь громоздкое и непредсказуемое)

• обслуживающий код сего декларативного обработчика лениво писать, может быть проблематично переиспользовать и уж тем более тяжело объединять с другими обработчиками в более сложные pipeline-ы

• сложно (если и вовсе возможно) динамически изменять заданную процедуру обработки

pip install convtools

Проблемы, описанные выше, наталкивают на мысли: "Было бы неплохо иметь возможность из питона задавать некие конверсии, которые можно цеплять друг за друга, а когда необходимая конверсия уже на руках, вызвать метод и получить узкоспециализированный код. Ну и хорошо бы заиметь group_by & join функциональность."

С этими мыслями и была создана библиотека convtools (conversion tools).

Приведу для примера несколько примитивов, которыми оперирует библиотека:

• c.item(key_or_index) - задает операцию для обращения по индексам и ключам (может принимать более одного и поддерживает default=... )

• c.attr(attr_name) - операция обращения к аттрибутам

• c.call_func(datetime.strptime, c.item("dt"), "%Y-%m-%d") - операция вызова функции (часть аргументов заранее инициализированы)

• c.iter(c.item("id")) - эквивалентно (item["id"] for item in input_data)

• c.this().call_method("replace", "abc", "cde") - эквивалентно input_data.replace("abc", "cde")

• c.item("object_list").pipe(any_other_conversion) - выхлоп любой конверсии можно направить в другую конверсию (в том числе group_by / join) или ф-ю

Перейдем, наконец, к: group_by

from convtools import conversion as c

input_data = [
    {"a": 5, "b": "foo"},
    {"a": 10, "b": "foo"},
    {"a": 10, "b": "bar"},
    {"a": 10, "b": "bar"},
    {"a": 20, "b": "bar"},
]
# Давайте сгруппируемся по "b" и найдем суммы и первые значения "a"
conv = (
    c.group_by(c.item("b"))
    .aggregate(
        {
            "b": c.item("b"),
            "a_first": c.ReduceFuncs.First(c.item("a")),
            "a_sum": c.ReduceFuncs.Sum(c.item("a")),
        } # этот dict можно собрать динамически, можно использовать
          # конверсии в качестве ключей, а можно и вовсе поменять на tuple
    )
    .gen_converter()  # в этом месте генерируется и компилируется код конверсии
    # если установить black и передать сюда debug=True, то в консоль выведет
    # форматированный код
)

assert conv(input_data) == [
    {'b': 'foo', 'a_first': 5, 'a_sum': 15},
    {'b': 'bar', 'a_first': 10, 'a_sum': 40}]
]

 def group_by__eu(data_):
    global labels_
    _none = v_rc
    signature_to_agg_data__eu = defaultdict(AggData__eu)

    for row__eu in data_:
        agg_data__eu = signature_to_agg_data__eu[row__eu["b"]]
        if agg_data__eu.v0 is _none:
            agg_data__eu.v0 = row__eu["a"]
            agg_data__eu.v1 = row__eu["a"] or 0
        else:
            agg_data__eu.v1 = agg_data__eu.v1 + (row__eu["a"] or 0)

    return [
        {
            "b": signature__eu,
            "a_first": (None if agg_data__eu.v0 is _none else agg_data__eu.v0),
            "a_sum": (0 if agg_data__eu.v1 is _none else agg_data__eu.v1),
        }
        for signature__eu, agg_data__eu in signature_to_agg_data__eu.items()
    ]

 def converter_6s(data_):
    global labels_
    return group_by__eu(data_)

Следующий на очереди: join

from convtools import conversion as c

collection_1 = [
    {"id": 1, "name": "Nick"},
    {"id": 2, "name": "Joash"},
    {"id": 3, "name": "Bob"},
]
collection_2 = [
    {"ID": "3", "age": 17, "country": "GB"},
    {"ID": "2", "age": 21, "country": "US"},
    {"ID": "1", "age": 18, "country": "CA"},
]
input_data = (collection_1, collection_2)

converter = (
    c.join(
        c.item(0),  # collection_1 (т.к. на входе tuple)
        c.item(1),  # collection_2
        c.and_(
            c.LEFT.item("id") == c.RIGHT.item("ID").as_type(int),
            c.RIGHT.item("age") >= 18,
        ),  # условия для join
        how="left",
    )  # на выходе генератор tuple-ов (left_item, right_item)
    .iter({  # итерируемся преобразуем каждый элемент в дикт
        "id": c.item(0, "id"),  # от левого возьмем id
        "name": c.item(0, "name"),  # name левого
        "age": c.item(1, "age", default=None),  # age правого
        "country": c.item(1, "country", default=None),  # country правого
    })
    .as_type(list)  # приведем к листу, т.к. до сих пор работали с генератором
    .gen_converter()  # создаём конвертер и храним где удобно
)

assert converter(input_data) == [
    {'id': 1, 'name': 'Nick', 'age': 18, 'country': 'CA'},
    {'id': 2, 'name': 'Joash', 'age': 21, 'country': 'US'},
    {'id': 3, 'name': 'Bob', 'age': None, 'country': None}]

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

• github convtools

• pypi convtools

• cheatsheet

Заключение

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

Пожалуйста, делитесь мыслями / пожеланиями. Спасибо!