Python кодогенерация на благо ETL — часть 2

Привет! В первой части я поделился мыслями, которые побудили к созданию python библиотеки convtools. Кратко о ней: предоставляет примитивы (конверсии), объединяя которые, можно описывать сложные конверсии для обработки данных. Конверсия генерирует узкоспециализированный код, компилирует его и возвращает функцию, решающую конкретную задачу.

В этот раз хотелось бы подробнее остановиться на двух моментах:

• как pipe позволяет повысить переиспользуемость кода

• новая часть библиотеки: Table — потоковый обработчик табличных данных

1) Pipes & code reuse

Надеюсь, что читать код с комментариями удобнее, чем оторванное от него косноязычное повествование, поэтому приступим:

from datetime import datetime from decimal import Decimal  from convtools import conversion as c   # Допустим нам нужно прочесть некий отчет и поработать с ним.  # описываем конверсии, необходимые при чтении отчета # сохраним то, как будут обрабатываться все входные строки c_str = c.this().call_method("strip") # то, как из обработанной строки получить Decimal (убиваем запятые - # разделители групп разрядов) c_decimal = c_str.call_method("replace", ",", "").as_type(Decimal) c_parse_date = c.call_func(     datetime.strptime, c.this(), "%Y-%m-%d" ).call_method("date") c_date = c_str.pipe(c_parse_date)  c_optional_date = c_str.pipe(c.if_(c.this(), с_parse_date, None)) c_optional_decimal = c_str.pipe(c.if_(c.this(), c_decimal, None))  # разменяем входные строки на понятные типы через (тут должен быть # Enum, но с диктом тоже красиво) c_product_type = c.naive(     {         "phone": 1,         "laptop": 2,     } ).item(c_str.call_method("lower"))   # теперь определим, что нас может интересовать в данных, направляя # каждое из полей в необходимую конверсию schema = {     "full_name": c.item("Full Name").pipe(c_str),     "sku": c.item("SKU").pipe(c_str),     "product_type": c.item("Product Type").pipe(c_product_type),     "earnings": c.item("Earnings").pipe(c_decimal),     "refunds": c.item("Refunds").pipe(c_optional_decimal),     "date": c.item("Date").pipe(c_date),     "date_of_birth": c.item("Date of Birth").pipe(c_optional_date), }  # теперь можем забыть о том, как что и откуда берется. # можно описать необходимую логику converter = (     # группируем по именам и датам     c.group_by(schema["full_name"], schema["date"])     .aggregate(         {             "full_name": schema["full_name"],             "date": schema["date"],             "date_of_birth": c.ReduceFuncs.First(schema["date_of_birth"]),             "total_earnings": c.ReduceFuncs.Sum(schema["earnings"]),         }     )     # отбросим "начинающих"     .filter(c.item("total_earnings") > 1000)     # сгруппируем по датам, соберем якобы полезную аналитику:     #  - кол-во не "начинающих"     #  - их средний заработок     #  - их медианная дата рождения     #  - dict: SKU -> сумма заработков (агрегация в агрегации)     .pipe(         c.group_by(c.item("date")).aggregate(             {                 "date": c.item("date"),                 "number_of_workers": c.ReduceFuncs.Count(),                 "average_earnings": c.ReduceFuncs.Average(                     c.item("total_earnings")                 ),                 "median_date_of_birth": c.ReduceFuncs.Median(                     c.item("date_of_birth")                 ),                 "earnings_by_sku": c.ReduceFuncs.DictSum(                     c.item("sku"),                     c.item("total_earnings"),                 ),             }         )     )     # debug=True - выведет в stdout сгенерированный код.     # если установлен black, он будет еще и отформатирован     .gen_converter() )  converter(input_data)

def pipe__cp(input__cp, _naive, _labels, _none):     return (         _naive["strptime_ox"](input__cp.strip(), _naive["v_7o"]).date()         if input__cp         else None     )   class AggData__cg:     __slots__ = ["v0", "v1"]      def __init__(self, _none=__none__):         self.v0 = _none         self.v1 = _none   def group_by__cg(data_, _naive, _labels, _none):     signature_to_agg_data__cg = defaultdict(AggData__cg)      for row__cg in data_:         agg_data__cg = signature_to_agg_data__cg[             (                 row__cg["Full Name"].strip(),                 _naive["strptime_ox"](                     row__cg["Date"].strip(), _naive["v_7o"]                 ).date(),             )         ]         if agg_data__cg.v0 is _none:             agg_data__cg.v0 = pipe__cp(                 row__cg["Date of Birth"].strip(), _naive, _labels, _none             )             agg_data__cg.v1 = (                 _naive["Decimal_4v"](                     row__cg["Earnings"].strip().replace(",", "")                 )                 or 0             )         else:             agg_data__cg.v1 = agg_data__cg.v1 + (                 _naive["Decimal_4v"](                     row__cg["Earnings"].strip().replace(",", "")                 )                 or 0             )      result_ = (         {             "full_name": signature__cg[0],             "date": signature__cg[1],             "date_of_birth": (                 None if agg_data__cg.v0 is _none else agg_data__cg.v0             ),             "total_earnings": (                 0 if agg_data__cg.v1 is _none else agg_data__cg.v1             ),         }         for signature__cg, agg_data__cg in signature_to_agg_data__cg.items()     )     filtered_result_ = [         i_4z for i_4z in result_ if (i_4z["total_earnings"] > 1000)     ]     return filtered_result_   class AggData__gp:     __slots__ = ["v0", "v1", "v2", "v3"]      def __init__(self, _none=__none__):         self.v0 = _none         self.v1 = _none         self.v2 = _none         self.v3 = _none   def group_by__gp(data_, _naive, _labels, _none):     signature_to_agg_data__gp = defaultdict(AggData__gp)      for row__gp in data_:         agg_data__gp = signature_to_agg_data__gp[row__gp["date"]]         if agg_data__gp.v0 is _none:             agg_data__gp.v0 = 1             agg_data__gp.v2 = [row__gp["date_of_birth"]]             agg_data__gp.v3 = _d = defaultdict(int)             _d[row__gp["sku"]] = row__gp["total_earnings"] or 0         else:             agg_data__gp.v0 = agg_data__gp.v0 + 1             agg_data__gp.v2.append(row__gp["date_of_birth"])             agg_data__gp.v3[row__gp["sku"]] = agg_data__gp.v3[                 row__gp["sku"]             ] + (row__gp["total_earnings"] or 0)         if agg_data__gp.v1 is _none:             if row__gp["total_earnings"] is not None:                 agg_data__gp.v1 = (1, row__gp["total_earnings"] * 1)         else:             if row__gp["total_earnings"] is not None:                 agg_data__gp.v1 = (                     agg_data__gp.v1[0] + 1,                     agg_data__gp.v1[1] + row__gp["total_earnings"] * 1,                 )      return [         {             "date": signature__gp,             "number_of_workers": (                 0 if agg_data__gp.v0 is _none else agg_data__gp.v0             ),             "average_earnings": (                 None                 if agg_data__gp.v1 is _none                 else (agg_data__gp.v1[1] / agg_data__gp.v1[0])             ),             "median_date_of_birth": (                 None                 if agg_data__gp.v2 is _none                 else _naive["median_59"](agg_data__gp.v2)             ),             "earnings_by_sku": (                 None if agg_data__gp.v3 is _none else (dict(agg_data__gp.v3))             ),         }         for signature__gp, agg_data__gp in signature_to_agg_data__gp.items()     ]   def converter_uk(data_):     global __naive_values__, __none__     _naive = __naive_values__     _none = __none__     _labels = {}     return _naive["group_by__gp"](         _naive["group_by__cg"](data_, _naive, _labels, _none),         _naive,         _labels,         _none,     ) 

Так как все конверсии являются выражениями, метод pipe позволяет комбинировать практически любые из них. pipe-ы разумно воспринимать как бесплатную с точки зрения производительности абстракцию, так как в случаях, когда не требуется добавление label-ов и входные данные не используются более одного раза, pipe не приводит к дополнительному вызову функции.

2) Table — потоковая обработка табличных данных

Время от времени у меня появлялась необходимость в работе с табличными данными (или схожими с ними). И меня всегда прельщала мысль «запечь» в код индексы колонок, чтобы обращения к ним были дешевыми. И вот, наконец, имея под рукой примитивы convtools, получилось реализовать задуманное, совершая малое количество телодвижений.

Логика работы: Table работает с итераторами лениво, вычитывая только первую строку, в которой может содержаться заголовок. Далее, опираясь на заголовок и запрошенные изменения, пишется код, который выполнит нужные преобразования над имеющимся итератором. Методы into_iter_rows и into_csv являются конечными.

from convtools.contrib.tables import Table from convtools import conversion as c  (     Table.from_csv(         "tests/csvs/ac.csv",         header=True,         dialect=Table.csv_dialect(delimiter="\t"),     )     # возьмем только колонки "a" и "c"     .take("a", "c")     # добавим две новых вычисляемых колонки     .update(B=c.col("a") + c.col("c"), D=c.call_func(abs, c.col("c")))     # переименуем     .rename({"a": "A"})     # отфильтруем строки     .filter(c.col("c") < 10)     # отбросим колонку "c"     .drop("c").into_csv("tests/csvs/out.csv") )   # объединим две таблицы по колонке "a" list(     Table.from_rows([(1, 2), (2, 3)], ["a", "b"])     .join(         Table.from_rows([(1, 3), (2, 4)], ["a", "c"]),         how="inner",         on=["a"],     )     .into_iter_rows(dict)  # поддерживаются list/tuple )

Также добавлены два изредка полезных метода:

""" Table 1      Table 2 | a | b |    | b | c | | 1 | 2 |    | 3 | 4 |  >>> table1.chain(table2, fill_value=" ")  Result: | a | b | c | | 1 | 2 |   | |   | 3 | 4 | """

""" Table 1      Table 2 | a | b |    | b | c | | 1 | 2 |    | 3 | 4 |              | 5 | 6 | >>> table1.zip(table2, fill_value=" ")  Result: | a | b | b | c | | 1 | 2 | 3 | 4 | |   |   | 5 | 6 | """ 

Заключение

Конечно, я не питаю иллюзий, что все резко бросятся использовать convtools, но всё равно захотелось поделиться. И даже не потому, что codegen добавляет гибкость, просто меня написание кода в таком стиле ставит на другие рельсы. С таким подходом я допускаю значительно меньше ошибок, становятся более очевидными стадии обработки данных, облегчается тестирование кода, т.к. работа ведется с чистыми «почти-функциями» + convtools поощряет работу с iterator-ами, что положительно сказывается на потребляемой памяти.

Всем успехов и, если что-нибудь придумаю или подскажете, до скорого!