Pythonにパイプライン演算子（のようなもの）+α

パイプライン演算子とは

g(f(e(d(c(b(a(x)))))))

これを

x |> a |> b |> c |> d |> e |> f |> g

ないし

x |> a
  |> b
  |> c
  |> d
  |> e
  |> f
  |> g

と書けるようにする演算子のこと。F#などで使われる。

最初の記法が処理の順番と読む順番が逆であるのに対し、パイプライン演算子を使うと処理の順番と読む順番を揃えられる点が便利。

これをPythonでも使いたいので実装を考えてみる。ついでに、パイプライン演算子を使う上で欲しい機能も実装していく。

関数として実装

とりあえず関数として実装する。

import functools as fts

def pipe(x, *fs):
    return fts.reduce(lambda x, y: y(x), fs, x)

def add2(x):
    return x + 2

def mul2(x):
    return x * 2

pipe(3,
     mul2,
     add2,
     add2)
# out:
# 10

演算子で実装

先人pipetoolsのようにちゃんと演算子を作る場合。

演算子として実装するのであれば既存の演算子をオーバーロードするしかない。オーバーロードするにはクラスにラップするしかない。

from functools import partial


pout = object()


class PipeBase:
    def __mod__(self, f):
        # 多重ディスパッチが欲しいところだ
        if f is pout:
            return self.out()
        else:
            return Pipe(self, f)


class Pipe(PipeBase):
    def __init__(self, previous, f):
        self._previous = previous
        self._f = f

    def out(self):
        return self._f(self._previous.out())


class FirstPipe(PipeBase):
    def __init__(self, first):
        self._first = first

    def out(self):
        return self._first


class InPipe:
    def out(self):
        return None

    def __mod__(self, first):
        if first is pout:
            return self.out()
        else:
            return FirstPipe(first)


pin = InPipe()

(pin %
    3 %
    mul2 %
    add2 %
    add2 %
    pout)
# out:
# 10

# 遅延評価もできる

lazy_pipe = (pin %
    3 %
    mul2 %
    add2 %
    add2)

lazy_pipe.out()
# out:
# 10

>とか|を使わなかった理由は後で説明する。

これはこれで悪くないと思う。複数行にわたるときにカッコを付けないといけないのと、pin/poutが面倒だが･･･

メソッド呼び出しもパイプに入れる

関数だけでなくメソッド呼び出しもしたいとしたらどうするか。

組み込み関数methodcallerを使うと、メソッド呼び出しを関数化できる。

from operator import methodcaller

pipe(3,
     add2,
     add2,
     mul2,
     add2,
     mul2,
     add2,
     mul2,
     mul2,
     mul2,
     str, # "272"
     methodcaller("count", "2"), # "272".count("2")
     add2
)
# out:
# 4

でもわざわざ文字列で書くのが面倒。なのでこんなクラスを作るともう少しすっきり書ける。

class MethodCaller:
    def __getattribute__(self, name):
        return partial(methodcaller, name)


call = MethodCaller()

pipe(3,
     add2,
     add2,
     mul2,
     add2,
     mul2,
     add2,
     mul2,
     mul2,
     mul2,
     str,
     call.count("2"),
     add2
)
# out:
# 4

属性呼び出しもパイプに入れる

属性呼び出しを関数にするならattrgetterを使う。

from collections import namedtuple
from operator import attrgetter


Point = namedtuple("Point", ["x", "y"])


pipe(Point(3, 2),
    attrgetter("y"),
    add2)
# out:
# 4

これもすっきりさせたかったらこんなクラスを定義すればいい。

class AttrGetter:
    def __getattribute__(self, name):
        return attrgetter(name)


get = AttrGetter()

pipe(Point(3, 2),
    get.y,
    add2)
# out:
# 4

部分適用もほしい

h(g(3, f(x)))

みたいなのを扱う場合はどうするか？

この場合は、関数gに「3だけ適用した」ような関数（これを部分適用という）を作った上で

x |> f |> (gに3だけ適用した関数) |> h

と書けばよい。

pythonの場合はpartialという部分適用のための関数が用意されているから、これを使えば書ける。

pipe(range(9),
    partial(filter, lambda x: x%2 == 0), # 偶数だけ取って
    partial(map, lambda x: x*2), # 二倍して
    partial(fts.reduce, lambda x, y: x+y)) # 足し算
# out:
# 40

partialっていちいち書くのが面倒ならpとか別名を付けてしまえばよい。実用的にはこれでいいと思う。

パイプ演算子とラッパークラスを定義している場合なら、部分適用のための演算子を追加することも出来る。

class Pipe(PipeBase):
    def __init__(self, previous, f):
        self._previous = previous
        self._f = f

    def out(self):
        return self._f(self._previous.out())

    def __truediv__(self, arg):
        return Pipe(self._previous, partial(self._f, arg))

(pin %
    range(9) %
    filter / (lambda x: x%2 == 0) %
    map / (lambda x: x*2) %
    fts.reduce / (lambda x, y: x+y) %
    pout)
# out:
# 40

演算子の優先順位の関係で、ラムダ式にいちいちカッコを付けなきゃいけないのが悲しいところだ。あと、キーワード引数への部分適用はできない。

ちなみに上のコードはパイプ演算子(%)と部分適用演算子(/)が同じ優先順位を持つことを利用している。>や|を使わなかったのは、同じ優先順位を持つ演算子がないからだ。

partial(filter, lambda x: x%2 ==0)じゃなくてfilter(lambda x: x%2 == 0)って書きたい！！！F#ちゃんとかHaskellちゃんは買ってもらってるもん！！うちはなんでだめなの！！みたいに駄々をこね出すとなかなか面倒なことになる。

メソッド呼び出しと同じやり方で解決しようと思うと関数の名前から関数を復元しないといけなくなるのだが、これはパイプ演算子を使っているコードの側の名前空間にアクセスしないといけない。黒魔術が必要。

from inspect import stack


class FirstPartialFunc:
    def __getattribute__(self, name):
        # スタックを遡り呼び出し元の名前空間を再現する
        frame = stack()[1].frame
        caller_namespace = {
            **frame.f_builtins,
            **frame.f_globals,
            **frame.f_locals
        }
        return PartialFunc(caller_namespace[name])


class PartialFunc:
    def __init__(self, namespace_or_func=None):
        self._namespace_or_func = namespace_or_func

    def __getattr__(self, name):
        # モジュールを介した呼び出しの場合もhandleできるようにする
        return PartialFunc(getattr(self._namespace_or_func, name))

    def __call__(self, *args, **kwargs):
        return partial(self._namespace_or_func, *args, **kwargs)


part = FirstPartialFunc()

pipe(range(9),
    part.filter(lambda x: x%2 == 0),
    part.map(lambda x: x*2),
    part.fts.reduce(lambda x, y: x+y))
# out:
# 40

多分これでいいと思うんだけどあんまり自信ない。

副作用も起こしたい

途中でprintしたい、みたいな話。

何にも考えないでpipeに突っ込むと戻り値がNoneになったりで計算が止まる。

値は素通しするような高階関数を定義してやればできる。

def do(f):
    def _do(x):
        f(x)
        return x
    return _do

pipe(range(9),
    partial(filter, lambda x: x%2 == 0),
    list,
    do(print),
    partial(map, lambda x: x*2),
    list,
    do(print),
    partial(fts.reduce, lambda x, y: x+y))
# [0, 2, 4, 6, 8]
# [0, 4, 8, 12, 16]
# out:
# 40

もちろん演算子でやることもできる。記号がいろいろ出てきて楽しい。

class Pipe(PipeBase):
    def __init__(self, previous, f):
        self._previous = previous
        self._f = f

    def out(self):
        return self._f(self._previous.out())

    def __truediv__(self, arg):
        return Pipe(self._previous, partial(self._f, arg))

    def __matmul__(self, f):
        return Pipe(self._previous, do(self._f)) % f

(pin %
    range(9) %
    filter / (lambda x: x%2 == 0) %
    list %
    print @
    map / (lambda x: x*2) %
    list %
    print @
    fts.reduce / (lambda x, y: x+y) %
    pout)
# [0, 2, 4, 6, 8]
# [0, 4, 8, 12, 16]
# out:
# 40

実用的には

pipe関数+partial+methodcaller+attrgetter+doで書くくらいならまあ特に変なコードでは無いと思う。Pythonicかは別として、コレクション処理を連続しなきゃいけないケースなんかではかなり簡潔にコードが書けるだろう。

演算子のオーバーロードとかマジックメソッドの実装とかスタックフレームのさかのぼりとかだんだん趣味の世界に入っていく。