Funktionales Pipelining

[Schard]

Moin,

ich mache gerade mal wieder Schabernack mit Python und habe list gesubclassed um eine Pipelining Syntax zum Filtern zu ermöglichen.
Ich denke nicht, dass dies aktuell wirklich nützlich ist; es dienst vornehmlich zur Unterhaltung und als Proof-of-Concept:

Code: Alles auswählen

#! /usr/bin/env python3

from typing import Any, Callable


class PipelineFilterableList(list):

    def __or__(self, predicate: Callable[[Any], bool]):
        return type(self)(filter(predicate, self))


def main():

    items = PipelineFilterableList([3, 9, 5, 13, 12, 30])
    filtered_items = items | (lambda x: x > 5) | (lambda x: x % 2 == 0)
    print(filtered_items)


if __name__ == '__main__':
    main()

Man kann es natürlich auch noch generischer machen:

Code: Alles auswählen

#! /usr/bin/env python3

from functools import partial
from typing import Callable, Iterable


class PipelinableList(list):

    def __or__(self, mutator: Callable[[Iterable], Iterable]):
        return type(self)(mutator(self))


def main():

    items = PipelinableList([3, 9, 5, 13, 12, 30])
    filtered_items = items | partial(filter, lambda x: x > 5) | partial(filter, lambda x: x % 2 == 0)
    squared_filtered_items = filtered_items | partial(map, lambda x: x ** 2)
    print(filtered_items)
    print(squared_filtered_items)


if __name__ == '__main__':
    main()

Um die Syntax wieder zu vereinfachen, kann man natürlich auch verschiedene Operatoren implementieren:

Code: Alles auswählen

#! /usr/bin/env python3

from typing import Any, Callable


class PipelinableList(list):

    def __or__(self, predicate: Callable[[Any], bool]):
        return type(self)(filter(predicate, self))

    def __matmul__(self, predicate: Callable[[Any], Any]):
        return type(self)(map(predicate, self))


def main():

    items = PipelinableList([3, 9, 5, 13, 12, 30])
    filtered_items = items | (lambda x: x > 5) | (lambda x: x % 2 == 0)
    squared_filtered_items = filtered_items @ (lambda x: x ** 2)
    print(filtered_items)
    print(squared_filtered_items)


if __name__ == '__main__':
    main()

Allerdings kann es da beim Chaining zu Problemen kommen, da @ stärker bindet als |.

[__blackjack__]

@Schard: Ich würde das nicht von `list` ableiten. Zum einen weil es dann viele Operationen gibt die auf einer `list` wieder eine `list` liefern würden, auf einer `PipelineFilterableList` aber keine `PipelineFilterableList` und damit die Pipeline unterbrechen. Slicing beispielsweise. Und bei Pipelines würde ich „lazy“ verhalten erwarten und nicht, dass da immer erst die kompletten Zwischenergebnisse in jedem Schritt realisiert werden.

Warum von List ableiten wenn eine Klasse die von nichts erbt und ein beliebiges „iterable“ wrappen kann und lazy funktioniert?

``>`` könnte man dann auch überladen und beispielsweise wenn eine Zeichenkette, ein `Path`-like oder ein Objekt mit einer `write()`-Methode übergeben wird dann das Ergebnis in eine Datei schreibt. Und bei einem aufrufbaren Objekt, das dann für jedes Element mit dem Element aufruft.

[kbr]

Zum Spielen ist das sicherlich einen Versuch wert, aber ich würde hier auch nicht in die Protokolle eingreifen, sonder eher eine Funktion im Stil von "map" oder "filter" schreiben. Oder konkret: einen Wrapper für "filter". Idee:

Code: Alles auswählen

def pipeline(functions, sequence):
    for function in functions:
        sequence = filter(function, sequence)
    return sequence
    
pipe = lambda x: x > 5, lambda x: x % 2 == 0
sequence = [3, 9, 5, 13, 12, 30]
result = pipeline(pipe, sequence)
print(list(result))

Wobei der Rückgabewert keine sequence, sondern eine filter-Instanz ist, quasi eine "lazy-sequence".

[Sirius3]

Hier meine Variante:

Code: Alles auswählen

class Mapper:
    def __init__(self, function):
        self.function = function
    
    def __ror__(self, iterable):
        return map(self.function, iterable)

class Filter:
    def __init__(self, function):
        self.function = function
    
    def __ror__(self, iterable):
        return filter(self.function, iterable)

print(list(range(10) | Mapper(lambda x: x*2) | Filter(lambda x: x > 5)))

[Schard]

@Sirius3
Das ist natürlich insofern besser, da, wie auch bereits von @__blackjack__ angesprochen, direkt die Iteratoren gechained werden, statt jedes Mal eine neue Liste zu konstruieren.