Go to Qiita Advent Calendar 2024 Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 3 years have passed since last update.

@robozushi10(robozushi10)

Observer Pattern の使用例

Last updated at Posted at 2021-09-26
関連記事のトップページ
[シリーズ] Python におけるデザインパターンの記録

概要

Python では Observer パターンを使用したことがなかったので、実装練習をしてみた.
このとき Python のパッケージ pypattyrn の Observer を使用することにした.

詳細

WEB や書籍には Observer パターンの実践例が多数あるが、
いずれも自分で Observer (監視者) と Observable (発信者) を実装するものが殆どであった.

私としては、(車輪の再開発による不要な劣化を回避したいので)
Python のサードパーティパッケージ pypattyrn の Observer を使用することにした.

なお、pypattyrn を使用した理由は、
Star 数が多く1 、且つ、ソースコードが読み解き易かったことが理由である.

実装内容

「従業員(Employee) の給与が更新されたら、監視対象(Observer)に通知する」
というプログラムを実装する.

これは、下記書籍(Ruby) にある Observer パターンを Python に置き換えたものである.
ただし、内容について細かく記すと書籍の無断転載になるので大まかに記している.

引用元情報 一言
書籍 -- Rubyによるデザインパターン (Russ Olsen 著) 原書(英文)
GitHub -- 著者 Russ Olsen 氏

クラスとファイル構成

pypattyrn が提供する Observable と Observer をそれぞれ継承して、
具象クラス Employee と TestObserver を作成する.2

image.png

pypattyrn の制約として以下があるので注意が必要である.

Observable (発信者) クラスのメンバ変数のうち、
アンダスコア1個 と 2個 (ダンダ) は Observer への通知対象外である.

上図 Employee のメンバ変数 (name, title, salary) を、
_name, _title, _salary にすると Observer へ通知されなくなってしまう.3

環境

- Python 3.9.4 を使用した
- 「pip install pypattyrn」で pypattyrn をインストールしておくこと

実装

ex6_std_observer.py

「発信者」のコードである.

給与は「self.salary」であり、self.salary への書き込み (Setter呼び出し) が発生したら、
メソッド notify を呼び出して Observer (監視者) に通知させるようにしている.

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import sys; sys.dont_write_bytecode = True
from pypattyrn.behavioral.observer import Observable, Observer

class Employee(Observable):
  def __init__(self, name, title, salary):
    super().__init__()
    self.name = name
    self.title = title
    self.salary = salary

  @property
  def salary(self):
    return self.__dict__['salary'] if 'salary' in self.__dict__ else None

  @salary.setter
  def salary(self, new_salary):
    print(f'[Employee] {self.title} ... {self.name}さんの給与の更新({self.salary} -> {new_salary}) を検知しました.')
    self.__dict__['salary'] = new_salary
    self.notify()

ex4_obs_class.py

Observer (監視者)側のコードである.

メソッド update の第2引数「**state」には、
アンダスコア1個と2個のメンバ変数を除いた Employee のフィールドが格納されている.

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import sys; sys.dont_write_bytecode = True
from pypattyrn.behavioral.observer import Observable, Observer
 
class TestObserver(Observer):
  update_state = None
 
  def update(self, **state):
    print(f'[TestObserver] TestObserver.update(self, **state) が呼び出されました')
    print(f'[TestObserver] 引数 **state の値を表示します')
    for k,v in state.items():
      print(f'[TestObserver] {k=:6} => {v=}')
    self.update_state = state
    print(f'[{self.update_state=}] done')

ex8_demo.py

上記 Observable と Observer を実行するプログラムである

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import sys; sys.dont_write_bytecode = True
from ex6_std_observer import *
from ex4_obs_class    import *

print('')
# Observable (いわゆる Subject == 発信者) を作成する
subject = Employee(name='fred', title='info', salary=1000)
# Observer (監視者) のインスタンスを作成する
observer1 = TestObserver()

# Observable (== Subject == 発信者) に通知先である Observer オブジェクトを登録する
subject.attach(observer1)
# 新たな給与を通知する
subject.salary = 1999

実行

上記 ex8_demo.py を実行すると次のような結果になる

[Employee] info ... fredさんの給与の更新(None -> 1000) を検知しました.
[Employee] info ... fredさんの給与の更新(1000 -> 1999) を検知しました.
[TestObserver] TestObserver.update(self, **state) が呼び出されました
[TestObserver] 引数 **state の値を表示します
[TestObserver] k=name   => v='fred'
[TestObserver] k=title  => v='info'
[TestObserver] k=salary => v=1999
[self.update_state={'name': 'fred', 'title': 'info', 'salary': 1999}] done

補足

単体テスト用コード

下記は「python3 -m unittest ex8_test.py」で実行可能である.

ex8_test.py

UnitTest を実行させたい場合のコード.

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import sys; sys.dont_write_bytecode = True
import unittest
from ex6_std_observer import *
from ex4_obs_class    import *

class EmployeeTest(unittest.TestCase):

  def test_1_none(self):
    print('')
    print('test_1_none を開始します')
    # Observable (いわゆる Subject == 発信者) を作成する
    subject = Employee(name='fred', title='info', salary=1000)
    # Observer (監視者) のインスタンスを作成する
    observer1 = TestObserver()

    # Observable (== Subject == 発信者) に通知先である Observer オブジェクトを登録する
    subject.attach(observer1)
    # 新たな給与を通知する
    subject.salary = 1999

  def test_2_new_observer(self):
    print('')
    print('test_2_new_observer を開始します')
    #-----------------------------------------------------------
    # Observable (いわゆる Subject == 発信者) を作成する
    # なお、この時点では Observer (監視者) を登録していないので通知はしない
    subject = Employee(name='fred', title='info', salary=1000)

    #-----------------------------------------------------------
    # Observer (監視者2) のインスタンスを作成する
    observer2 = TestObserver()
    # 作成したインスタンスが TestObserver であるかを検証する
    self.assertIsInstance(observer2, TestObserver)
    # 作成したインスタンスが Observer を継承しているか検証する
    self.assertIsInstance(observer2, Observer)
    # Observable (== Subject == 発信者) に通知先である Observer オブジェクトを登録する
    subject.attach(observer2)

    #-----------------------------------------------------------
    # Observer (監視者3) のインスタンスを作成する
    observer3 = TestObserver()
    # 作成したインスタンスが TestObserver であるかを検証する
    self.assertIsInstance(observer3, TestObserver)
    # 作成したインスタンスが Observer を継承しているか検証する
    self.assertIsInstance(observer3, Observer)
    # Observable (== Subject == 発信者) に通知先である Observer オブジェクトを登録する
    subject.attach(observer3)

    #-----------------------------------------------------------
    # 発信者側に Observer オブジェクト 2つが登録できたかチェックする.
    # (ここのテストは pypattyrn/behavioral/observer.py の内部を知っている前提である)
    assert subject._observers is not None
    assert observer2 in subject._observers
    assert observer3 in subject._observers
    # 新たな給与を通知する
    subject.salary = 1234

実行結果

$ python -m unittest ex8_test.py

test_1_none を開始します
[Employee] info ... fredさんの給与の更新(None -> 1000) を検知しました.
[Employee] info ... fredさんの給与の更新(1000 -> 1999) を検知しました.
[TestObserver] TestObserver.update(self, **state) が呼び出されました
[TestObserver] 引数 **state の値を表示します
[TestObserver] k=name   => v='fred'
[TestObserver] k=title  => v='info'
[TestObserver] k=salary => v=1999
[self.update_state={'name': 'fred', 'title': 'info', 'salary': 1999}] done
.
test_2_new_observer を開始します
[Employee] info ... fredさんの給与の更新(None -> 1000) を検知しました.
[Employee] info ... fredさんの給与の更新(1000 -> 1234) を検知しました.
[TestObserver] TestObserver.update(self, **state) が呼び出されました
[TestObserver] 引数 **state の値を表示します
[TestObserver] k=name   => v='fred'
[TestObserver] k=title  => v='info'
[TestObserver] k=salary => v=1234
[self.update_state={'name': 'fred', 'title': 'info', 'salary': 1234}] done
[TestObserver] TestObserver.update(self, **state) が呼び出されました
[TestObserver] 引数 **state の値を表示します
[TestObserver] k=name   => v='fred'
[TestObserver] k=title  => v='info'
[TestObserver] k=salary => v=1234
[self.update_state={'name': 'fred', 'title': 'info', 'salary': 1234}] done
.
----------------------------------------------------------------------
Ran 2 tests in 0.001s

OK

　

以上

　

  1. 最終更新から 5年ほど経過しているが、「枯れているため」と判断した ↩

  2. UML 図は正確では無いかも知れない ↩

  3. Observer への不要な更新を避けるための工夫なのだと思う ↩

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?