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-принцип и не делая конечный код медленнее, т.к. в нем не будет лишних условий, циклов и вызовов ф-й.

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