CUBE SUGAR CONTAINER

技術系のこと書きます。

Python: 関数合成できる API を作ってみる

Python macOS

今回は普通の Python では満足できなくなってしまった人向けの話題。 dfply や pipe といった一部のパッケージで採用されているパイプ処理や関数合成できる API を作る一つのやり方について。

使った環境は次の通り。

$ sw_vers                  
ProductName:    Mac OS X
ProductVersion: 10.14.6
BuildVersion:   18G1012
$ python -V
Python 3.7.5

もくじ

カッコ以外で評価されるオブジェクトを作る

通常の Python では、関数やメソッドはカッコを使ってオブジェクトを評価する。 しかし、クラスを定義するとき特殊メソッドを使って演算子オーバーライドすることで、その枠に収まらないオブジェクトが作れる。

例えば、以下のように関数をラップするクラスを定義する。 特殊メソッドの __rrshift__() は、自身の「左辺」にある右ビットシフト演算子が評価されるときに呼び出される。 なお、別に右ビットシフト演算子を使う必然性はないので、別の演算子をオーバーライドしても構わない。

>>> class Pipe:
...     """関数をラップするクラス"""
...     def __init__(self, f):
...         # インスタンス化するとき関数を受け取る
...         self.f = f
...     def __rrshift__(self, other):
...         # 自身の左辺にある右ビットシフト演算子を評価するとき関数を実行する
...         return self.f(other)
...

これを使って、例えば値を二乗するオブジェクトを作ってみよう。

>>> pow = Pipe(lambda x: x ** 2)

このオブジェクトに右ビットシフト演算子を使って値を渡すと、その内容が二乗される。

>>> 10 >> pow
100

他の関数も定義してつなげるとメソッドチェーンっぽいことができる。

>>> double = Pipe(lambda x: x * 2)
>>> 10 >> pow >> double
200

ただ、このままだと関数と関数だけをつないだときに例外になってしまう。 この場合は、左辺にあるオブジェクトの右辺にある右ビットシフト演算子が先に評価されているため。

>>> pow >> double
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: unsupported operand type(s) for >>: 'Pipe' and 'Pipe'

関数合成できるオブジェクトを作る

そこで、先ほどのクラスに手を加える。 以下のように、自身の右辺に渡された処理を自身のリストにキューイングしておけるようにする。

>>> import copy
>>> class Pipe:
...     """関数合成に対応したクラス"""
...     def __init__(self, f):
...         self.f = f
...         # 適用したい一連の関数を記録しておくリスト
...         self.pipes = []
...     def __rshift__(self, other):
...         """自身の右辺にある右ビットシフト演算子を評価したときに呼ばれる特殊メソッド"""
...         # 自身をコピーしておく
...         copied_self = copy.deepcopy(self)
...         # コピーした内容のリストに適用したい処理をキューイングする
...         copied_self.pipes.append(other)
...         # コピーした自身を返す
...         return copied_self
...     def __rrshift__(self, other):
...         """自身の左辺にある右ビットシフト演算子を評価したときに呼ばれる特殊メソッド"""
...         # まずは自身の関数を適用する
...         result = self.f(other)
...         # キューイングされていた関数を順番に適用していく
...         for pipe in self.pipes:
...             result = pipe.__rrshift__(result)
...         # 最終的な結果を返す
...         return result
...

こうすると、関数同士をつないだ場合にもオブジェクトが返るようになる。

>>> pow = Pipe(lambda x: x ** 2)
>>> double = Pipe(lambda x: x * 2)
>>> pow >> double
<__main__.Pipe object at 0x102090950>

上記を変数に保存しておいて値を適用すると、ちゃんと本来の意図通りにチェーンされた結果が返ってくる。

>>> pow_double = pow >> double
>>> 10 >> pow_double
200

自身をディープコピーする理由について

ちなみに、先ほど右辺にある右ビットシフト演算子が評価されるときに自身のオブジェクトをディープコピーしていた。 もし、ディープコピーしないとどうなるだろうか。 実際にやってみよう。

>>> class Pipe:
...     def __init__(self, f):
...         self.f = f
...         self.pipes = []
...     def __rshift__(self, other):
...         # 自身をコピーせずに関数をキューイングする場合
...         self.pipes.append(other)
...         return self
...     def __rrshift__(self, other):
...         result = self.f(other)
...         for pipe in self.pipes:
...             result = pipe.__rrshift__(result)
...         return result
...

コピーしない場合でも、ちゃんと Pipe オブジェクトは返ってくる。

>>> pow = Pipe(lambda x: x ** 2)
>>> double = Pipe(lambda x: x * 2)
>>> pow >> double
<__main__.Pipe object at 0x102090a50>

しかし、右辺の右ビットシフト演算子が評価された時点で適用する処理のリストにキューイングされてしまう。

>>> pow.pipes
[<__main__.Pipe object at 0x102090b50>]

つまり、元のオブジェクトを変更してしまう。 本来なら二乗してほしいだけのオブジェクトで二倍も同時にされてしまうことになる。

>>> 10 >> pow
200

デコレータとして使う

ちなみに、ここまで作ってきたクラスはクラスデコレータとして使うこともできる。

ようするに、次のように関数をクラスでデコレートできる。

>>> @Pipe
... def triple(x):
...     return x * 3
... 
>>> @Pipe
... def half(x):
...     return x // 2
... 
>>> 10 >> triple >> half
15

デコレータの詳細については以下を参照のこと。

blog.amedama.jp

引数を受け取れるようにする

次に、適用される関数に引数を渡したくなる。 この場合、__call__() メソッドを実装して関数に引数を渡す形でオブジェクトを作り直すようにすると良い。

>>> class Pipe:
...     def __init__(self, f):
...         self.f = f
...         self.pipes = []
...     def __rshift__(self, other):
...         copied_self = copy.deepcopy(self)
...         copied_self.pipes.append(other)
...         return copied_self
...     def __rrshift__(self, other):
...         result = self.f(other)
...         for pipe in self.pipes:
...             result = pipe.__rrshift__(result)
...         return result
...     def __call__(self, *args, **kwargs):
...         """オブジェクトが実行されたときに呼ばれる特殊メソッド"""
...         # 実行されたときの引数を関数に渡すようにしたオブジェクトを返す
...         return Pipe(lambda x: self.f(x, *args, **kwargs))
...

例えば、掛ける数を引数にした掛け算を実装してみよう。

>>> @Pipe
... def multiply(x, n):
...     return x * n
...

これは、次のように使うことができる。

>>> 10 >> multiply(2) >> multiply(3)
60

おもしろいね。

参考プロジェクト

github.com

github.com

スマートPythonプログラミング: Pythonのより良い書き方を学ぶ

スマートPythonプログラミング: Pythonのより良い書き方を学ぶ