Пайплайны и частичное применения функций, зачем это в Python

Одно из главных достоинств Python — его выразительность. Функциональные средства языка позволяют лаконично описывать преобразования над данными. На мой взгляд в Python не хватает некоторых инструментов, которые помогли бы удобнее описывать преобразования данных и дополнить функциональную составляющую языка, в частности "пайплайны функций" и их частичное применение. Поэтому в этом посте я лью воду о возможности и необходимости данных средств с экспериментами по их реализации. Пришёл во многом за критикой. Приятного чтения!

Кратко о ФП в Python и почему не хватает пайплайнов на примере

В Python из базовых средств есть довольно удобные map(), reduce(), filter(), лямбда-функции, итераторы и генераторы. Малознакомым с этим всем советую данную статью. В целом это оно всё позволяет быстро и естественно описывать преобразования над списками, кортежами, и тд. Очень часто(у меня и знакомых питонистов) то, что получается однострочник — по сути набор последовательных преобразований, фильтраций, например:
Kata с CodeWars: Найти

Задачка довольно простая, к сожалению(но к счастью для этого поста), решений лучше чем в лоб нет.

Моё решение:

def sum_dig_pow(a, b): # range(a, b + 1) will be studied by the function
    powered_sum = lambda x: sum([v**(i+1) for i,v in enumerate(map(lambda x: int(x), list(str(x))))])
    return [i for i in range(a,b+1) if powered_sum(i)==i]

С использованием средств ФП как есть получается скобочный ад "изнутри наружу". Это мог бы исправить пайплайн.

Пайплайны функций

Под сим я подразумеваю такое в идеальном случае (оператор "|" — личное предпочтение):

# f3(f2(f1(x)))
f1 | f2 | f3 >> x
pipeline = f1 | f2 | f3 
pipeline(x)
pipeline2 = f4 | f5
pipeline3 = pipeline | pipeline2 | f6
...

Тогда powered_sum может стать(код не рабочий):

powered_sum = str | list | map(lambda x: int(x), *args) | enumerate | [v**(i+1) for i,v in *args] | sum

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

from copy import deepcopy

class CreatePipeline:
    def __init__(self, data=None):
        self.stack = []
        if data is not None:
            self.args = data

    def __or__(self, f):
        new = deepcopy(self)
        new.stack.append(f)
        return new

    def __rshift__(self, v):
        new = deepcopy(self)
        new.args = v
        return new

    def call_logic(self, *args):
        for f in self.stack:
            if type(args) is tuple:
                args = f(*args)
            else:
                args = f(args)
        return args

    def __call__(self, *args):
        if 'args' in self.__dict__:
            return self.call_logic(self.args)
        else:
            return self.call_logic(*args)

Естественно, это один большой костыль, состряпанный ради интереса, даже без kwargs, хотя в похожих случаях и не так важно.

pipe = CreatePipeline()
powered_sum = pipe | str | list | (lambda l: map(lambda x: int(x), l)) | enumerate | (lambda e: [v**(i+1) for i,v in e]) | sum

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

Частичное применение функций

Рассмотрим на примере простейшей функции(код не рабочий):

def f_partitial (x,y,z):
    return x+y+z
v = f_partial(1,2)
# type(v) = что-нибудь частично применённая функция f_partial, оставшиеся аргументы: ['z']
print(v(3))
# Эквивалент
print(f_partial(1,2,3))

Такая возможность была бы полезна для пайпа и другого разного(насколько фантазии хватит). Тогда пример с учётом имеющейся реализации pipe может стать таким:

powered_sum = pipe | str | list | map(lambda x: int(x)) | enumerate | (lambda e: [v**(i+1) for i,v in e]) | sum
# map будет вызван ещё раз со вторым аргументом
# map(lambda x: int(x))(данные) при вызове

map(lambda x: int(x)) в пайплайне выглядит более лаконично в целом и в терминах последовательных преобразований данных.
Кривенькая неполная реализация на уровне языка:

from inspect import getfullargspec
from copy import deepcopy

class CreatePartFunction:
    def __init__(self, f):
        self.f = f
        self.values = []

    def __call__(self, *args):
        args_f = getfullargspec(self.f)[0]
        if len(args) + len(self.values) < len(args_f):
            new = deepcopy(self)
            new.values = new.values + list(args)
            return new
        elif len(self.values) + len(args) == len(args_f):
            return self.f(*tuple(self.values + list(args)))

Реализация примера с учётом данного костыля дополнения:

# костыль для обхода поломки inspect над встроенным map
m = lambda f, l: map(f, l)
# создаём частично применяемую функцию на основе обычной питоньей
pmap = CreatePartFunction(m)
powered_sum = pipe | str | list | pmap(lambda x: int(x)) | enumerate | (lambda e: [v**(i+1) for i,v in e]) | sum

При более чем двойном вызове в строке(что в целом не особо нужно), придётся уже расставлять скобки, потому что питон подумает, что вызывается аргумент, то есть:

def f (x,y,z):
    return x+y+z
f = CreatePartFunction(f)
# работает
print(f(1,2,3))
# работает
print(f(1,2)(3))
print(f(1)(2,3))
# не работает
# 2(3) - int не callable
print(f(1)(2)(3))
# работает
print((f(1)(2))(3))

Итоги

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