Haskellの合成演算子 . をPythonで〜関数合成によるCollection処理の実装

メソッドチェインを利用した関数合成の実現案的な。当然ですが実用性汎用性ともに一切ございません。

事前に用意しておく必要があるので、Haskellでの.みたいな汎用性は一切ないですウワー。ただ、あらかじめ決められたオブジェクトへの関数の合成に応用することは出来るし、案外有用なのでは？

というわけで、世の中に無数にあるLINQ to ObjectのPython移植をつくります。何番煎じだとか言わないでやめて実装がちょっと違うの！
つまり、世の中のLINQ移植の実装をみると、そのほとんどが基本的にラッパーオブジェクトを生成して返すメソッドチェインで実現しています。今回の記事では、関数合成と要素への適用という形で実装してみます。

この時点で、先ほどのcomposableみたいに合成を行う以外に、事前に用意した要素への適用を行う（toListとか、firstとかanyのような IEnumerable -> Uみたいなことを行う系）decoratorも必要になりそうですね。

完成したものがこちらになります。

# -*- coding: utf-8 -*-
from __future__ import (print_function, division, absolute_import, unicode_literals, )

from six import get_function_code


def _compose(f_t_u, f_u_r):
    return lambda t: f_u_r(f_t_u(t))


class NotIterableError(Exception):
    pass


class Stream(object):
    __slots__ = ('value', 'xs', )

    def __init__(self, xs):
        self.xs = xs
        self.value = lambda ys: ys

    def bind(self, l_T_to_l_U):
        return _compose(self.value, l_T_to_l_U)

    def map(self, l_T_to_l_U):
        self.value = _compose(self.value, l_T_to_l_U)
        return self

    def __iter__(self):
        return iter(self.evaluate())

    def evaluate(self):
        try:
            return self.value(self.xs)
        except TypeError as e:
            return Empty(NotIterableError(str(e)))


class Empty(Stream):
    __slots__ = ('value', 'xs', 'error', )

    def __init__(self, error=None):
        self.error = error
        super(Stream, self).__init__(None)

    def map(self, l_T_to_l_U):
        return self

    def evaluate(self):
        return []

    def __repr__(self):
        return super(Stream, self).__repr__() + " reason => " + str(type(self.error)) + ": " + str(self.error)


def dispatch_stream(original_query):
    '''
    decorator to dispatch the function should be chaining method of masala.Stream

    :type original_query: AnyObjects -> iter
    '''
    func_name = get_function_code(original_query).co_name

    def _method_chaining_base(self, *args, **kw):
        return self.map(
            lambda xs: original_query(
                xs,
                *args,
                **kw
            )
        )
    setattr(Stream, func_name, _method_chaining_base)


def delete_dispatchedmethods(names):
    for name in names:
        delattr(Stream, name)


def endpoint_of_stream(original_query):
    '''
    decorator to dispatch the function should be end of chaining method of masala.Stream

    :type original_query: AnyObjects -> T
    '''
    func_name = get_function_code(original_query).co_name

    def _evaluatable_method(self, *args, **kw):
        return self.map(lambda xs: original_query(xs, *args, **kw)).evaluate()
    setattr(Stream, func_name, _evaluatable_method)

ちょっと頑張って、iteratorへの処理に失敗した場合やなんかにEmptyというオブジェクトを返すようにしてみました。Maybeモナドみたいな感じですね。LINQはモナドなので仕方ないね。
なお、Python2, 3両対応にするために一部sixというサードパーティを利用しています。便利です。

ちゃんと動作するか試してみましょう。

from six import PY2
if PY2:
    from itertools import (
        imap,
        ifilter,
    )
else:
    imap = map
    ifilter = filter


@dispatch_stream
def select(xs, y_from_x):
    from itertools import imap
    return imap(y_from_x, xs)


@dispatch_stream
def where(xs, predicate):
    return ifilter(predicate, xs)


@endpoint_of_stream
def to_list(xs):
    return list(xs)


if __name__ == '__main__':
    ret = Stream(range(10)).select(lambda x: x * 2).where(lambda x: x % 3 == 0).to_list()
    assert ret == [0, 6, 12, 18, ]