pyramid_bubblingというものを作った

pyramid python

pyramid_bubblingというものを作った。

まだpypiにはあげていない。 https://github.com/podhmo/pyramid_bubbling

何これ？

いわゆるイベントのバブリングを実装したかった。ルーティング イベントとも言われるかもしれない。 オブジェクト間の関連を木構造で表した時に末端の葉から発火されたイベントが実行されながら根まで届くというような振る舞いを実装したい。

使い方

使い方は大雑把に２通り。

  • classが直接的に関連を持つような形
  • 関連をinterface(ZCA)越しに持つような形

前者の方が振る舞いは単純で、後者の方が柔軟。

usecase

以下のような構造のものが在るとする。

Document -> Area -> Button

Documentが根でButtonが葉。 ここでButtonをクリックした時、クリックイベントが親であるArea,Documentにも伝搬されて欲しい。

classが直接的に関連を持つような形の場合

# -*- coding:utf-8 -*-

from pyramid_bubbling import (
    bubbling_attribute,
    Bubbling,
    Accessor, 
    Stop
)

class Document(object):
    def __init__(self, name):
        self.name = name

    def on_click(self, subject, result):
        result.append(subject.name)

    def on_tap(self, subject, result):
        result.append(("tap", subject.name))

class Area(object):
    def __init__(self, name, parent=None):
        self.name = name
        self.parent = parent

    @bubbling_attribute(Document)
    def __parent__(self):
        return self.parent

    def on_click(self, subject, result):
        result.append(subject.name)

    def on_tap(self, subject, result):
        result.append(("tap", subject.name))

    def on_mouse_over(self, subject, result):
        result.append(subject.name)
        return Stop

class Button(object):
    def __init__(self, name, parent=None):
        self.name = name
        self.parent = parent

    @bubbling_attribute(Area)
    def __parent__(self):
        return self.parent

    def on_click(self, subject, result):
        result.append(subject.name)

    def on_tap(self, subject, result):
        result.append(("tap", subject.name))

    def on_mouse_over(self, subject, result):
        result.append(subject.name)

    @bubbling_attribute(Document)
    def __document__(self):
        return self.parent.parent

document = Document("doc")
content_area = Area("content", document)
ok_button = Button("ok", content_area)

bubbling = Bubbling()

## click event
result = []
bubbling.fire(ok_button, "click", result)

assert result == ["ok", "content", "doc"]

## tap event
result = []
bubbling.fire(ok_button, "tap", result)

assert result == [('tap', 'ok'), ('tap', 'content'), ('tap', 'doc')]

デフォルトではCoC的に暗黙のルールを元に振る舞いをする。 以下の様なルールで動く

  • 子から親の関連は parent 属性へのアクセスで行われる
  • fire()で渡された文字列に"on_"を前置したメソッドが呼ばれる
  • イベントのcallbackの戻り値がStopの時伝搬が途中で打ち切られる。

stop

pyramid_bubbling.Stopが返されたイベントは途中で伝搬が打ち切られる

## mouse over event
result = []
bubbling.fire(ok_button, "mouse_over", result)

assert result == ["ok", "content"]

ルールの変更

子から親への関連を取得する操作は、Bubblingオブジェクトに渡す引数を変える事で変更できる。

## get path via __document__
document_bubbling = Bubbling(Accessor("__document__"))
result = []
document_bubbling.fire(ok_button, "tap", result)

assert result == [('tap', 'ok'), ('tap', 'doc')]

イベントの取得方法は、Accessorオブジェクトのget_notifyを変更することで変えられる。

## イベントのcallbackを保持するdict
event_table = {}

class AccessorUseDict(Accessor):
    Table = event_table
    def get_notify(self, subject, case):
        clsname = subject.__class__.__name__.lower()
        try:
            return self.Table["{}_{}".format(clsname, case)]
        except KeyError:
            return self.Table[case]


def on_click(subject, result):
    result.append(("click", subject.name))

def on_click_document(subject, result):
    result.append(("document", "click", subject.name))

event_table["click"] = on_click
event_table["document_click"] = on_click_document


document_bubbling2 = Bubbling(AccessorUseDict("__document__"))
result = []
document_bubbling2.fire(ok_button, "click", result)

assert result == [('click', 'ok'), ('click', 'content')]

関連をinterface(ZCA)越しに持つような形

次回に続く。