Update 2023.11.17 2017.05.05

Python デザインパターン サンプルコード Chain of Responsibility
Mark Summerfield『実践 Python 3』デザインパターンのサンプルコード
Python3(3.11)で動くソースコード(.pyファイル .ipynbファイル)あります
「anaconda3」on .py「PyCharm」.ipynb「Jupyter Notebook」

著作権の問題があるので,本に書いてないことだけを解説します。つまり,視点を変えて解説します。

◆◆Chain of Responsibility パターンの使われる場面◆◆

以下は,同じサイトの別記事について説明されている。
https://yamakatsusan.web.fc2.com/hp_software/pythonpattern18.html
『Chain of Responsibility パターン 結城 浩「Java言語で学ぶデザインパターン入門」をPython化』

GoFのC++サンプルは GUI のオンラインヘルプです。解決策が複数ある場合に解決策を提示するために問題側を解析してしまうと,問題側と解決側の関係が密になりすぎる傾向になります。そうならないように,問題側をいじらないというのがChain of Responsibility パターンです。

結城氏は「責任のたらい回し」と言っていますが,むしろ,組織で責任を持って対応するというのが正しいと思われます。

例えば,役所に相談に来た人に対して,窓口の人が相談内容を聴き出すのも対応の1つですが,そういう対応ではなく Chain of Responsibility パターンでは,複数の部署が順に対応するという解決策をとります。ということでこのパターンは問題を選ばないと変な対応になってしまうかもしれません。

Java サンプルは全く実用的でない例になっていて,問題側は問題に番号を付けるだけ,解決側は問題の番号を見て解決可能かを示すだけという内容になっています。Chain of Responsibility パターンのサンプルとしてこれで十分なのかもしれません。

(2023-11-17)Python3.11で動作確認済み


◆◆Chain of Responsibility パターンとは◆◆

GoFによれば,Chain of Responsibility パターンの目的は, 「1つ以上のオブジェクトに要求を処理する機会を与えることにより,要求を送信するオブジェクトと受信するオブジェクトの結合を避ける。受信する複数のオブジェクトをチェーン状につなぎ,あるオブジェクトがその要求を処理するまで,そのチェーンに沿って要求を渡していく。」

GoFによれば,Chain Of Responsibility パターンは,次のような場合に使うことができる。
・要求を処理するオブジェクトの候補が複数存在し,最終的にどのオブジェクトが担当するのかは,前もってわからない場合。担当オブジェクトは自動的に決められる。
・受け手を明確にせずに,複数あるオブジェクトの1つに対して要求を発行したい場合。
・要求を処理することができるオブジェクトの集合が動的に明確化される場合。

GoFによれば,Chain of Responsibility パターンの関連するパターンは次のようなものである。
Composite パターン:Chain Of Responsibility パターンは,しばしば Composite パターンとともに適用される。その場合,component の親オブジェクトを successor にすることができる。

◆◆Chain of Responsibility パターンのサンプルのクラス構成◆◆

ソースコードは巻末にあり,ソースファイルはこのWebの左上隅にありダウンロードできます。
【eventhandler1.py】
       NullHandler
      (handle)
       ↑              ↑             ↑               ↑
    DebugHandler    MouseHandler    KeyHandler    TimerHandler
    (handle)         (handle)        (handle)      (handle)
class NullHandler は,スーパークラスです。
class DebugHandler は,デバッグのときに使います。
class MouseHandler, KeyHandler, TimerHandler は,イベントに対応する処理をします。

【eventhandler2.py】
    def debug_handler
    def mouse_handler
    def key_handler
    def timer_handler
eventhandler1.py のクラスに対応するコルーチンがあります。

◆◆Chain of Responsibility パターンの実装◆◆

【eventhandler1.py】

Chain of Responsibility パターンは,巻頭にも書きましたが,送信側と受信側,問題側と解決側のような区分けができる場合,両者の関係が,1対1,1対多であった場合,その関係を直接書いてしまうと関係が密になり過ぎます。

そこで,送信側や問題側で何が起ころうとも,受信側や解決側は,すべての処理を鎖状に繋げてしまい,たらい回しするというものです。処理が終わったら,次に回すのを止めます。

サンプルには,いわゆる送信側や問題側には,3つのイベントと処理終了を意味するものがあります(Event モジュール 17行目)。
  MOUSE
  KEYPRESS
  TIMER
  TERMINATE
これらのイベントが起こると,送信されるものは(Event モジュール 47行目),
  "Button"
  "Key"
  "Timer"
  "Terminate"
であります。これらは,eventhandler1.py の次のクラスに対応しています(TERMINATEを除く)。
  class MouseHandler
  class KeyHandler
  class TimerHandler
これらのクラスは,コンストラクタを持っていません。そこで,スーパークラスである
  class NullHandler
のコンストラクタが使われます。それには,後任を指定できることになっています。

また,上の3つのクラスの同名のインスタンスメソッド
  def handle
では,受信したイベント名称が自分であれば,処理をして,そうでなければ,スーパークラスの同名メソッドを呼び出し,さらに後任の同名メソッドを呼び出します。

処理側を鎖状に繋げるのは,18行目の
  handler1 = TimerHandler(KeyHandler(MouseHandler(NullHandler())))
であります。後任は,コンストラクタの引数に書かれます。

もう1つの鎖状に繋げているのは,27行目の
  handler2 = DebugHandler(handler1)
であり,46行目の
  class DebugHandler
には,コンストラクタがあり,そのインスタンスメソッド
  def handle
では,何も処理はしないで,常に,後任の名前を print して,その後任を呼び出します。debug のつもりだと思います。

サンプル出力は次のようになります。書式に重要な意味はありません。

Handler Chain #1
Press:   Key Shift+b
Click:   Button 3 (584, 427)
Press:   Key i
Press:   Key Shift+m
Press:   Key Shift+m
Press:   Key t
Press:   Key Ctrl+Shift+f
Click:   Button 3 (255, 91)
Click:   Button 2 (49, 112)
Press:   Key Ctrl+p
Click:   Button 2 (189, 125)
Press:   Key k
Click:   Button 2 (239, 286)

Handler Chain #2 (debugging)
*DEBUG*: Timer 0
Timeout: Timer 0
*DEBUG*: Timer 1
Timeout: Timer 1
*DEBUG*: Button 2 (105, 144)
Click:   Button 2 (105, 144)
*DEBUG*: Timer 2
Timeout: Timer 2
*DEBUG*: Key r
Press:   Key r
*DEBUG*: Key Ctrl+s
Press:   Key Ctrl+s
*DEBUG*: Key z
Press:   Key z
*DEBUG*: Key Shift+j
Press:   Key Shift+j
*DEBUG*: Key Shift+j
Press:   Key Shift+j
*DEBUG*: Key y
Press:   Key y
*DEBUG*: Key x
Press:   Key x

【eventhandler2.py】

eventhandler1.py と違う点は,クラスがすべてグローバル関数になっていて,さらに,デコレータにより,コルーチンになっています。

yield を持つ関数は,普通,ジェネレータと言います。ジェネレータは,yield で一時停止しますが,for in ループで,イテレートされた(繰り返し呼ばれた)とき,値を pull します。

一方,コルーチンは,無限ループの中に,yield を持ち,そこで一時停止しているのだが,24, 32行目の
  pipeline.send(event)
によって,値を push されます。

コルーチンを鎖状に繋げるのは,19, 27行目の
  pipeline = key_handler(mouse_handler(timer_handler()))
  pipeline = debug_handler(pipeline)
であります。

サンプル出力は,eventhandler1.py と同じです。

◆◆ソースコード◆◆

このWebページの左上隅からダウンロードできます。

ソースファイルは4つです。
・eventhandler1.py;Chain of Responsibility パターンのPythonサンプル(その1)
・eventhandler2.py;Chain of Responsibility パターンのPythonサンプル(その2)
・Event.py;eventhandler1.py, eventhandler2.py がインポートする
・Qtrac.py;eventhandler2.pyがインポートする

【eventhandler1.py】
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
#!/usr/bin/env python3
# Copyright c 2012-13 Qtrac Ltd. All rights reserved.
# This program or module is free software: you can redistribute it
# and/or modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version. It is provided for
# educational purposes and is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.

import sys
import Event


def main():
    print("Handler Chain #1")
    handler1 = TimerHandler(KeyHandler(MouseHandler(NullHandler())))
    # Could pass None or nothing instead of the NullHandler
    while True:
        event = Event.next()
        if event.kind == Event.TERMINATE:
            break
        handler1.handle(event)

    print("\nHandler Chain #2 (debugging)")
    handler2 = DebugHandler(handler1)
    while True:
        event = Event.next()
        if event.kind == Event.TERMINATE:
            break
        handler2.handle(event)


class NullHandler:

    def __init__(self, successor=None):
        self.__successor = successor


    def handle(self, event):
        if self.__successor is not None:
            self.__successor.handle(event)


class DebugHandler(NullHandler):

    def __init__(self, successor=None, file=sys.stdout):
        super().__init__(successor)
        self.__file = file


    def handle(self, event):
        self.__file.write("*DEBUG*: {}\n".format(event))
        super().handle(event)


class MouseHandler(NullHandler):

    def handle(self, event):
        if event.kind == Event.MOUSE:
            print("Click:   {}".format(event))
        else:
            super().handle(event)


class KeyHandler(NullHandler):

    def handle(self, event):
        if event.kind == Event.KEYPRESS:
            print("Press:   {}".format(event))
        else:
            super().handle(event)


class TimerHandler(NullHandler):

    def handle(self, event):
        if event.kind == Event.TIMER:
            print("Timeout: {}".format(event))
        else:
            super().handle(event)


if __name__ == "__main__":
    main()

【eventhandler2.py】
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
#!/usr/bin/env python3
# Copyright c 2012-13 Qtrac Ltd. All rights reserved.
# This program or module is free software: you can redistribute it
# and/or modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version. It is provided for
# educational purposes and is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.

import sys
import Event
from Qtrac import coroutine


def main():
    print("Handler Chain #1")
    pipeline = key_handler(mouse_handler(timer_handler()))
    while True:
        event = Event.next()
        if event.kind == Event.TERMINATE:
            break
        pipeline.send(event)

    print("\nHandler Chain #2 (debugging)")
    pipeline = debug_handler(pipeline)
    while True:
        event = Event.next()
        if event.kind == Event.TERMINATE:
            break
        pipeline.send(event)


@coroutine
def debug_handler(successor, file=sys.stdout):
    while True:
        event = (yield)
        file.write("*DEBUG*: {}\n".format(event))
        successor.send(event)


@coroutine
def mouse_handler(successor=None):
    while True:
        event = (yield)
        if event.kind == Event.MOUSE:
            print("Click:   {}".format(event))
        elif successor is not None:
            successor.send(event)


@coroutine
def key_handler(successor=None):
    while True:
        event = (yield)
        if event.kind == Event.KEYPRESS:
            print("Press:   {}".format(event))
        elif successor is not None:
            successor.send(event)


@coroutine
def timer_handler(successor=None):
    while True:
        event = (yield)
        if event.kind == Event.TIMER:
            print("Timeout: {}".format(event))
        elif successor is not None:
            successor.send(event)


if __name__ == "__main__":
    main()

【Event.py】
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
#!/usr/bin/env python3
# Copyright c 2012-13 Qtrac Ltd. All rights reserved.
# This program or module is free software: you can redistribute it
# and/or modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version. It is provided for
# educational purposes and is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.

import random
import string

random.seed(917) # Don't want random for regression tests

MOUSE, KEYPRESS, TIMER, TERMINATE = ("MOUSE", "KEYPRESS", "TIMER",
        "TERMINATE")


def next():
    kinds = ([MOUSE] * 7) + ([KEYPRESS] * 11) + ([TIMER] * 5) + [TERMINATE]
    kind = random.choice(kinds)
    if kind == MOUSE:
        return Event(kind, button=random.randint(1, 3),
                x=random.randint(0, 640), y=random.randint(0, 480))
    elif kind == KEYPRESS:
        return Event(kind, ctrl=random.randint(1, 7) == 1,
                shift=random.randint(1, 5) == 1,
                key=random.choice(string.ascii_lowercase))
    return Event(kind) # TIMER or TERMINATE


class Event:

    TimerId = 0

    def __init__(self, kind, **kwargs):
        assert kind in {MOUSE, KEYPRESS, TIMER, TERMINATE}
        self.kind = kind
        self.kwargs = kwargs
        if self.kind == TIMER:
            self.kwargs["id"] = Event.TimerId
            Event.TimerId += 1


    def __str__(self):
        if self.kind == MOUSE:
            return "Button {} ({}, {})".format(
                    self.kwargs.get("button", 1), self.kwargs.get("x", -1),
                    self.kwargs.get("y", -1))
        elif self.kind == KEYPRESS:
            return "Key {}{}{}".format(
                    "Ctrl+" if self.kwargs.get("ctrl", False) else "",
                    "Shift+" if self.kwargs.get("shift", False) else "",
                    self.kwargs.get("key", ""))
        elif self.kind == TIMER:
            return "Timer {}".format(self.kwargs.get("id", -1))
        elif self.kind == TERMINATE:
            return "Terminate"

【Qtrac.py】
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#!/usr/bin/env python3
# Copyright c 2012-13 Qtrac Ltd. All rights reserved.
# This program or module is free software: you can redistribute it
# and/or modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version. It is provided for
# educational purposes and is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.

import abc
import collections
import errno
import functools
import os
import sys


def coroutine(function):
    @functools.wraps(function)
    def wrapper(*args, **kwargs):
        generator = function(*args, **kwargs)
        next(generator)
        return generator
    return wrapper

以上

トップページに戻る
inserted by FC2 system