一、创建型模式
1、工厂方法 Factory
工厂方法是一种创建型设计模式，其在父类中提供一个创建对象的方法，允许子类决定实例化对象的类型
制造业是一个国家工业经济发展的重要支柱，而工厂则是其根基所在。程序设计中的工厂类往往是对对象构造、实例化、初始化过程的封装，而工厂方法则可以升华为一种设计模式，它对工厂制造方法进行接口规范化，以允许子类工厂决定具体制造哪类产品的实例，最终降低系统耦合，使系统的可维护性、可扩展性等得到提升。
委托专门的函数/方法来创建实例，实例 -> 类 -> 类工厂。
from typing import Dict
from typing import Protocol
from typing import Type
class Localizer(Protocol):
def localize(self, msg: str) -> str:
pass
class GreekLocalizer:
"""A simple localizer a la gettext"""
def __init__(self) -> None:
self.translations = {"dog": "σκύλος", "cat": "γάτα"}
def localize(self, msg: str) -> str:
"""We'll punt if we don't have a translation"""
return self.translations.get(msg, msg)
class EnglishLocalizer:
"""Simply echoes the message"""
def localize(self, msg: str) -> str:
return msg
def get_localizer(language: str = "English") -> Localizer:
"""Factory"""
localizers: Dict[str, Type[Localizer]] = {
"English": EnglishLocalizer,
"Greek": GreekLocalizer,
}
return localizers[language]()
def main():
"""
# Create our localizers
>>> e, g = get_localizer(language="English"), get_localizer(language="Greek")
# Localize some text
>>> for msg in "dog parrot cat bear".split():
... print(e.localize(msg), g.localize(msg))
dog σκύλος
parrot parrot
cat γάτα
bear bear
"""
if __name__ == "__main__":
import doctest
doctest.testmod()2、抽象工厂 Abstract_factory
简单来说就是把一些具有相同方法的类再进行封装，抽象共同的方法以供调用，是工厂方法的进阶版本。
实例 -> 类 -> 类工厂 -> 抽象工厂。
import random
from typing import Type
class Pet:
def __init__(self, name: str) -> None:
self.name = name
def speak(self) -> None:
raise NotImplementedError
def __str__(self) -> str:
raise NotImplementedError
class Dog(Pet):
def speak(self) -> None:
print("woof")
def __str__(self) -> str:
return f"Dog<{self.name}>"
class Cat(Pet):
def speak(self) -> None:
print("meow")
def __str__(self) -> str:
return f"Cat<{self.name}>"
class PetShop:
"""A pet shop"""
def __init__(self, animal_factory: Type[Pet]) -> None:
"""pet_factory is our abstract factory. We can set it at will."""
self.pet_factory = animal_factory
def buy_pet(self, name: str) -> Pet:
"""Creates and shows a pet using the abstract factory"""
pet = self.pet_factory(name)
print(f"Here is your lovely {pet}")
return pet
# Additional factories:
# Create a random animal
def random_animal(name: str) -> Pet:
"""Let's be dynamic!"""
return random.choice([Dog, Cat])(name)
# Show pets with various factories
def main() -> None:
"""
# A Shop that sells only cats
>>> cat_shop = PetShop(Cat)
>>> pet = cat_shop.buy_pet("Lucy")
Here is your lovely Cat<Lucy>
>>> pet.speak()
meow
# A shop that sells random animals
>>> shop = PetShop(random_animal)
>>> for name in ["Max", "Jack", "Buddy"]:
... pet = shop.buy_pet(name)
... pet.speak()
... print("=" * 20)
Here is your lovely Cat<Max>
meow
====================
Here is your lovely Dog<Jack>
woof
====================
Here is your lovely Dog<Buddy>
woof
====================
"""
if __name__ == "__main__":
random.seed(1234) # for deterministic doctest outputs
shop = PetShop(random_animal)
import doctest
doctest.testmod()3、惰性初始化 Lazy evaluation
类的某个属性来自于一个复杂的耗时的计算，但并不是每次都会调用。通过lazy evaluation模式，可以使该值只在真正需要读取的时候才进行一次计算。
这个Python里可以使用@property实现，就是当调用的时候才生成。
import functools
class lazy_property:
def __init__(self, function):
self.function = function
functools.update_wrapper(self, function)
def __get__(self, obj, type_):
if obj is None:
return self
val = self.function(obj)
obj.__dict__[self.function.__name__] = val
return val
def lazy_property2(fn):
"""
A lazy property decorator.
The function decorated is called the first time to retrieve the result and
then that calculated result is used the next time you access the value.
"""
attr = "_lazy__" + fn.__name__
@property
def _lazy_property(self):
if not hasattr(self, attr):
setattr(self, attr, fn(self))
return getattr(self, attr)
return _lazy_property
class Person:
def __init__(self, name, occupation):
self.name = name
self.occupation = occupation
self.call_count2 = 0
@lazy_property
def relatives(self):
# Get all relatives, let's assume that it costs much time.
relatives = "Many relatives."
return relatives
@lazy_property2
def parents(self):
self.call_count2 += 1
return "Father and mother"
def main():
"""
>>> Jhon = Person('Jhon', 'Coder')
>>> Jhon.name
'Jhon'
>>> Jhon.occupation
'Coder'
# Before we access `relatives`
>>> sorted(Jhon.__dict__.items())
[('call_count2', 0), ('name', 'Jhon'), ('occupation', 'Coder')]
>>> Jhon.relatives
'Many relatives.'
# After we've accessed `relatives`
>>> sorted(Jhon.__dict__.items())
[('call_count2', 0), ..., ('relatives', 'Many relatives.')]
>>> Jhon.parents
'Father and mother'
>>> sorted(Jhon.__dict__.items())
[('_lazy__parents', 'Father and mother'), ('call_count2', 1), ..., ('relatives', 'Many relatives.')]
>>> Jhon.parents
'Father and mother'
>>> Jhon.call_count2
1
"""
if __name__ == "__main__":
import doctest
doctest.testmod(optionflags=doctest.ELLIPSIS)4、生成器 Builder
Builder模式主要用于构建一个复杂的对象，但这个对象构建的算法是稳定的，对象中的各个部分经常变化。Builder模式主要在于应对复杂对象各个部分的频繁需求变动。但是难以应对算法的需求变动。这点一定要注意，如果用错了，会带来很多不必要的麻烦。
重点是将复杂对象的建造过程抽象出来（抽象类别），使这个抽象过程的不同实现方法可以构造出不同表现（属性）的对象。
简单的说：子对象变化较频繁，对算法相对稳定。
# Abstract Building
class Building:
def __init__(self) -> None:
self.build_floor()
self.build_size()
def build_floor(self):
raise NotImplementedError
def build_size(self):
raise NotImplementedError
def __repr__(self) -> str:
return "Floor: {0.floor} | Size: {0.size}".format(self)
# Concrete Buildings
class House(Building):
def build_floor(self) -> None:
self.floor = "One"
def build_size(self) -> None:
self.size = "Big"
class Flat(Building):
def build_floor(self) -> None:
self.floor = "More than One"
def build_size(self) -> None:
self.size = "Small"
# In some very complex cases, it might be desirable to pull out the building
# logic into another function (or a method on another class), rather than being
# in the base class '__init__'. (This leaves you in the strange situation where
# a concrete class does not have a useful constructor)
class ComplexBuilding:
def __repr__(self) -> str:
return "Floor: {0.floor} | Size: {0.size}".format(self)
class ComplexHouse(ComplexBuilding):
def build_floor(self) -> None:
self.floor = "One"
def build_size(self) -> None:
self.size = "Big and fancy"
def construct_building(cls) -> Building:
building = cls()
building.build_floor()
building.build_size()
return building
def main():
"""
>>> house = House()
>>> house
Floor: One | Size: Big
>>> flat = Flat()
>>> flat
Floor: More than One | Size: Small
# Using an external constructor function:
>>> complex_house = construct_building(ComplexHouse)
>>> complex_house
Floor: One | Size: Big and fancy
"""
if __name__ == "__main__":
import doctest
doctest.testmod()5、单例模式 Borg
希望一个类只有一个实例，在实例之间具有共享状态的实例。
方法一：使用__new__实现单例模式 > 使用__new__实现单例模式，具体我对__new__的理解可以点这里。
class SingleTon(object):
_instance = {}
def __new__(cls, *args, **kwargs):
if cls not in cls._instance:
cls._instance[cls] = super(SingleTon, cls).__new__(cls, *args, **kwargs)
# print cls._instance
return cls._instance[cls]
class MyClass(SingleTon):
class_val = 22
def __init__(self, val):
self.val = val
def obj_fun(self):
print self.val, 'obj_fun'
@staticmethod
def static_fun():
print 'staticmethod'
@classmethod
def class_fun(cls):
print cls.class_val, 'classmethod'
if __name__ == '__main__':
a = MyClass(1)
b = MyClass(2)
print a is b # True
print id(a), id(b) # 4367665424 4367665424
# 类型验证
print type(a) # <class '__main__.MyClass'>
print type(b) # <class '__main__.MyClass'>方法二：使用装饰器实现单例模式。
from functools import wraps
def single_ton(cls):
_instance = {}
@wraps(cls)
def single(*args, **kwargs):
if cls not in _instance:
_instance[cls] = cls(*args, **kwargs)
return _instance[cls]
return single
@single_ton
class SingleTon(object):
val = 123
def __init__(self, a):
self.a = a
if __name__ == '__main__':
s = SingleTon(1)
t = SingleTon(2)
print s is t
print s.a, t.a
print s.val, t.val方法三：使用模块实现单例模式。
可以使用模块创建单例模式，然后在其他模块中导入该单例，这个需要所有人遵守导入规则，不然就没法实现单例了。
# use_module.py
class SingleTon(object):
def __init__(self, val):
self.val = val
single = SingleTon(2)
# test_module.py
from use_module import single
a = single
b = single
print a.val, b.val
print a is b
a.val = 233
print a.val, b.val方法四：使用metaclass实现单例模式。
6、原型模式 Prototype
定义：原型实例指定创建对象的种类，并通过拷贝这些原型创建新的对象。属于创建型模式。原型模式和单例模式是互斥的。
通俗地说，原型模式就是快速构建对象的方法总结，简单工厂将get/set方法封装在某个方法中，JDK提供的实现Cloneable接口。
调用者不需要知道任何创建细节，不调用构造函数创建，直接拷贝。简化了产生对象的繁琐的过程，不需要通过构造方法去构造。
适用场景：
- 类初始化消耗资源较多时；
- new产生对象的过程非常繁琐（数据准备，访问权限等）；
- 构造函数比较复杂，不走构造函数，直接复制；
- 循环体中需要生产大量对象时；
特点是通过复制一个已经存在的实例来返回新的实例,而不是新建实例。
多用于创建复杂的或者耗时的实例,因为这种情况下，复制一个已经存在的实例使程序运行更高效，或者创建值相等,只是命名不一样的同类数据。
from __future__ import annotations
from typing import Any
class Prototype:
def __init__(self, value: str = "default", **attrs: Any) -> None:
self.value = value
self.__dict__.update(attrs)
def clone(self, **attrs: Any) -> Prototype:
"""Clone a prototype and update inner attributes dictionary"""
# Python in Practice, Mark Summerfield
# copy.deepcopy can be used instead of next line.
obj = self.__class__(**self.__dict__)
obj.__dict__.update(attrs)
return obj
class PrototypeDispatcher:
def __init__(self):
self._objects = {}
def get_objects(self) -> dict[str, Prototype]:
"""Get all objects"""
return self._objects
def register_object(self, name: str, obj: Prototype) -> None:
"""Register an object"""
self._objects[name] = obj
def unregister_object(self, name: str) -> None:
"""Unregister an object"""
del self._objects[name]
def main() -> None:
"""
>>> dispatcher = PrototypeDispatcher()
>>> prototype = Prototype()
>>> d = prototype.clone()
>>> a = prototype.clone(value='a-value', category='a')
>>> b = a.clone(value='b-value', is_checked=True)
>>> dispatcher.register_object('objecta', a)
>>> dispatcher.register_object('objectb', b)
>>> dispatcher.register_object('default', d)
>>> [{n: p.value} for n, p in dispatcher.get_objects().items()]
[{'objecta': 'a-value'}, {'objectb': 'b-value'}, {'default': 'default'}]
>>> print(b.category, b.is_checked)
a True
"""
if __name__ == "__main__":
import doctest
doctest.testmod()7、对象池 Object pool
一个对象池是一组已经初始化过且可以使用的对象，而可以不用在有需求时创建和销毁对象。池的用户可以从池子中取得对象，对其进行操作处理，并在不需要时归还给池子而非销毁 而不是销毁它。
在Python内部实现了对象池技术，例如像小整型这样的数据引用非常多，创建销毁都会消耗时间，所以保存在对象池里，减少开销。
class ObjectPool:
def __init__(self, queue, auto_get=False):
self._queue = queue
self.item = self._queue.get() if auto_get else None
def __enter__(self):
if self.item is None:
self.item = self._queue.get()
return self.item
def __exit__(self, Type, value, traceback):
if self.item is not None:
self._queue.put(self.item)
self.item = None
def __del__(self):
if self.item is not None:
self._queue.put(self.item)
self.item = None
def main():
"""
>>> import queue
>>> def test_object(queue):
... pool = ObjectPool(queue, True)
... print('Inside func: {}'.format(pool.item))
>>> sample_queue = queue.Queue()
>>> sample_queue.put('yam')
>>> with ObjectPool(sample_queue) as obj:
... print('Inside with: {}'.format(obj))
Inside with: yam
>>> print('Outside with: {}'.format(sample_queue.get()))
Outside with: yam
>>> sample_queue.put('sam')
>>> test_object(sample_queue)
Inside func: sam
>>> print('Outside func: {}'.format(sample_queue.get()))
Outside func: sam
if not sample_queue.empty():
print(sample_queue.get())
"""
if __name__ == "__main__":
import doctest
doctest.testmod()二、结构型模式
1、装饰器模型 Decorator
装饰器（Decorator）模式的定义：指在不改变现有对象结构的情况下，动态地给该对象增加一些职责（即增加其额外功能）的模式，它属于对象结构型模式。
装饰器模式的主要优点有：
- 装饰器是继承的有力补充，比继承灵活，在不改变原有对象的情况下，动态的给一个对象扩展功能，即插即用
- 通过使用不用装饰类及这些装饰类的排列组合，可以实现不同效果
- 装饰器模式完全遵守开闭原则
其主要缺点是：装饰器模式会增加许多子类，过度使用会增加程序得复杂性。
class TextTag:
"""Represents a base text tag"""
def __init__(self, text: str) -> None:
self._text = text
def render(self) -> str:
return self._text
class BoldWrapper(TextTag):
"""Wraps a tag in <b>"""
def __init__(self, wrapped: TextTag) -> None:
self._wrapped = wrapped
def render(self) -> str:
return f"<b>{self._wrapped.render()}</b>"
class ItalicWrapper(TextTag):
"""Wraps a tag in <i>"""
def __init__(self, wrapped: TextTag) -> None:
self._wrapped = wrapped
def render(self) -> str:
return f"<i>{self._wrapped.render()}</i>"
def main():
"""
>>> simple_hello = TextTag("hello, world!")
>>> special_hello = ItalicWrapper(BoldWrapper(simple_hello))
>>> print("before:", simple_hello.render())
before: hello, world!
>>> print("after:", special_hello.render())
after: <i><b>hello, world!</b></i>
"""
if __name__ == "__main__":
import doctest
doctest.testmod()2、三层架构模式 3-tier
三层架构指：
- 表现层(presentation layer): 也称UI层，程序运行的入口，如果程序包括界面，界面就放在这一层。
- 业务逻辑层(business logic layer)：王城程序的业务逻辑，对数据访问层进行调用，将从数据访问层中获取的数据反馈给表现层
- 数据层(data)：直接操作数据库。
其目的是为了实现“高内聚低耦合”的状态。高内聚是指，一个模块有相关性很强的代码组成，且只负责一项任务，即单一责任原则。低耦合是指：模块之间尽量独立。
from typing import Dict, KeysView, Optional, Union
class Data:
"""Data Store Class"""
products = {
"milk": {"price": 1.50, "quantity": 10},
"eggs": {"price": 0.20, "quantity": 100},
"cheese": {"price": 2.00, "quantity": 10},
}
def __get__(self, obj, klas):
print("(Fetching from Data Store)")
return {"products": self.products}
class BusinessLogic:
"""Business logic holding data store instances"""
data = Data()
def product_list(self) -> KeysView[str]:
return self.data["products"].keys()
def product_information(
self, product: str
) -> Optional[Dict[str, Union[int, float]]]:
return self.data["products"].get(product, None)
class Ui:
"""UI interaction class"""
def __init__(self) -> None:
self.business_logic = BusinessLogic()
def get_product_list(self) -> None:
print("PRODUCT LIST:")
for product in self.business_logic.product_list():
print(product)
print("")
def get_product_information(self, product: str) -> None:
product_info = self.business_logic.product_information(product)
if product_info:
print("PRODUCT INFORMATION:")
print(
f"Name: {product.title()}, "
+ f"Price: {product_info.get('price', 0):.2f}, "
+ f"Quantity: {product_info.get('quantity', 0):}"
)
else:
print(f"That product '{product}' does not exist in the records")
def main():
"""
>>> ui = Ui()
>>> ui.get_product_list()
PRODUCT LIST:
(Fetching from Data Store)
milk
eggs
cheese
<BLANKLINE>
>>> ui.get_product_information("cheese")
(Fetching from Data Store)
PRODUCT INFORMATION:
Name: Cheese, Price: 2.00, Quantity: 10
>>> ui.get_product_information("eggs")
(Fetching from Data Store)
PRODUCT INFORMATION:
Name: Eggs, Price: 0.20, Quantity: 100
>>> ui.get_product_information("milk")
(Fetching from Data Store)
PRODUCT INFORMATION:
Name: Milk, Price: 1.50, Quantity: 10
>>> ui.get_product_information("arepas")
(Fetching from Data Store)
That product 'arepas' does not exist in the records
"""
if __name__ == "__main__":
import doctest
doctest.testmod()3、适配器模式 Adapter
在不改变现有程序的情况下，通过创建一个适配器，去扩展我们所需的程序。
from typing import Callable, TypeVar
T = TypeVar("T")
class Dog:
def __init__(self) -> None:
self.name = "Dog"
def bark(self) -> str:
return "woof!"
class Cat:
def __init__(self) -> None:
self.name = "Cat"
def meow(self) -> str:
return "meow!"
class Human:
def __init__(self) -> None:
self.name = "Human"
def speak(self) -> str:
return "'hello'"
class Car:
def __init__(self) -> None:
self.name = "Car"
def make_noise(self, octane_level: int) -> str:
return f"vroom{'!' * octane_level}"
class Adapter:
"""Adapts an object by replacing methods.
Usage
------
dog = Dog()
dog = Adapter(dog, make_noise=dog.bark)
"""
def __init__(self, obj: T, **adapted_methods: Callable):
"""We set the adapted methods in the object's dict."""
self.obj = obj
self.__dict__.update(adapted_methods)
def __getattr__(self, attr):
"""All non-adapted calls are passed to the object."""
return getattr(self.obj, attr)
def original_dict(self):
"""Print original object dict."""
return self.obj.__dict__
def main():
"""
>>> objects = []
>>> dog = Dog()
>>> print(dog.__dict__)
{'name': 'Dog'}
>>> objects.append(Adapter(dog, make_noise=dog.bark))
>>> objects[0].__dict__['obj'], objects[0].__dict__['make_noise']
(<...Dog object at 0x...>, <bound method Dog.bark of <...Dog object at 0x...>>)
>>> print(objects[0].original_dict())
{'name': 'Dog'}
>>> cat = Cat()
>>> objects.append(Adapter(cat, make_noise=cat.meow))
>>> human = Human()
>>> objects.append(Adapter(human, make_noise=human.speak))
>>> car = Car()
>>> objects.append(Adapter(car, make_noise=lambda: car.make_noise(3)))
>>> for obj in objects:
... print("A {0} goes {1}".format(obj.name, obj.make_noise()))
A Dog goes woof!
A Cat goes meow!
A Human goes 'hello'
A Car goes vroom!!!
"""
if __name__ == "__main__":
import doctest
doctest.testmod(optionflags=doctest.ELLIPSIS)4、桥接模式 Bridge
桥接（Bridge）模式的定义如下：将抽象与实现分离，使它们可以独立变化。它是用组合关系代替继承关系来实现，从而降低了抽象和实现这两个可变维度的耦合度。
# ConcreteImplementor 1/2
class DrawingAPI1:
def draw_circle(self, x, y, radius):
print(f"API1.circle at {x}:{y} radius {radius}")
# ConcreteImplementor 2/2
class DrawingAPI2:
def draw_circle(self, x, y, radius):
print(f"API2.circle at {x}:{y} radius {radius}")
# Refined Abstraction
class CircleShape:
def __init__(self, x, y, radius, drawing_api):
self._x = x
self._y = y
self._radius = radius
self._drawing_api = drawing_api
# low-level i.e. Implementation specific
def draw(self):
self._drawing_api.draw_circle(self._x, self._y, self._radius)
# high-level i.e. Abstraction specific
def scale(self, pct):
self._radius *= pct
def main():
"""
>>> shapes = (CircleShape(1, 2, 3, DrawingAPI1()), CircleShape(5, 7, 11, DrawingAPI2()))
>>> for shape in shapes:
... shape.scale(2.5)
... shape.draw()
API1.circle at 1:2 radius 7.5
API2.circle at 5:7 radius 27.5
"""
if __name__ == "__main__":
import doctest
doctest.testmod()5、组合模式 Composite
将对象组合成树形结构以表示“部分-整体”的层次结构。Composite使得用户对单个对象和组合对象的使用具有一致性。
from abc import ABC, abstractmethod
from typing import List
class Graphic(ABC):
@abstractmethod
def render(self) -> None:
raise NotImplementedError("You should implement this!")
class CompositeGraphic(Graphic):
def __init__(self) -> None:
self.graphics: List[Graphic] = []
def render(self) -> None:
for graphic in self.graphics:
graphic.render()
def add(self, graphic: Graphic) -> None:
self.graphics.append(graphic)
def remove(self, graphic: Graphic) -> None:
self.graphics.remove(graphic)
class Ellipse(Graphic):
def __init__(self, name: str) -> None:
self.name = name
def render(self) -> None:
print(f"Ellipse: {self.name}")
def main():
"""
>>> ellipse1 = Ellipse("1")
>>> ellipse2 = Ellipse("2")
>>> ellipse3 = Ellipse("3")
>>> ellipse4 = Ellipse("4")
>>> graphic1 = CompositeGraphic()
>>> graphic2 = CompositeGraphic()
>>> graphic1.add(ellipse1)
>>> graphic1.add(ellipse2)
>>> graphic1.add(ellipse3)
>>> graphic2.add(ellipse4)
>>> graphic = CompositeGraphic()
>>> graphic.add(graphic1)
>>> graphic.add(graphic2)
>>> graphic.render()
Ellipse: 1
Ellipse: 2
Ellipse: 3
Ellipse: 4
"""
if __name__ == "__main__":
import doctest
doctest.testmod()6、外观模式 Facade
外观模式，我们通过外观的包装，使应用程序只能看到外观对象，而不会看到具体的细节对象，这样无疑会降低应用程序的复杂度，并且提高了程序的可维护性。
# Complex computer parts
class CPU:
"""
Simple CPU representation.
"""
def freeze(self) -> None:
print("Freezing processor.")
def jump(self, position: str) -> None:
print("Jumping to:", position)
def execute(self) -> None:
print("Executing.")
class Memory:
"""
Simple memory representation.
"""
def load(self, position: str, data: str) -> None:
print(f"Loading from {position} data: '{data}'.")
class SolidStateDrive:
"""
Simple solid state drive representation.
"""
def read(self, lba: str, size: str) -> str:
return f"Some data from sector {lba} with size {size}"
class ComputerFacade:
"""
Represents a facade for various computer parts.
"""
def __init__(self):
self.cpu = CPU()
self.memory = Memory()
self.ssd = SolidStateDrive()
def start(self):
self.cpu.freeze()
self.memory.load("0x00", self.ssd.read("100", "1024"))
self.cpu.jump("0x00")
self.cpu.execute()
def main():
"""
>>> computer_facade = ComputerFacade()
>>> computer_facade.start()
Freezing processor.
Loading from 0x00 data: 'Some data from sector 100 with size 1024'.
Jumping to: 0x00
Executing.
"""
if __name__ == "__main__":
import doctest
doctest.testmod(optionflags=doctest.ELLIPSIS)7、享元模式 Flyweight
当我们需要使用大量细粒度对象时，如果每个都需要创建实例，可能会浪费大量空间。如果这些对象有共同的内部状态，我们可以考虑使用Flyweight模式。
import weakref
class Card:
"""The Flyweight"""
# Could be a simple dict.
# With WeakValueDictionary garbage collection can reclaim the object
# when there are no other references to it.
_pool: weakref.WeakValueDictionary = weakref.WeakValueDictionary()
def __new__(cls, value, suit):
# If the object exists in the pool - just return it
obj = cls._pool.get(value + suit)
# otherwise - create new one (and add it to the pool)
if obj is None:
obj = object.__new__(Card)
cls._pool[value + suit] = obj
# This row does the part we usually see in `__init__`
obj.value, obj.suit = value, suit
return obj
# If you uncomment `__init__` and comment-out `__new__` -
# Card becomes normal (non-flyweight).
# def __init__(self, value, suit):
# self.value, self.suit = value, suit
def __repr__(self):
return f"<Card: {self.value}{self.suit}>"
def main():
"""
>>> c1 = Card('9', 'h')
>>> c2 = Card('9', 'h')
>>> c1, c2
(<Card: 9h>, <Card: 9h>)
>>> c1 == c2
True
>>> c1 is c2
True
>>> c1.new_attr = 'temp'
>>> c3 = Card('9', 'h')
>>> hasattr(c3, 'new_attr')
True
>>> Card._pool.clear()
>>> c4 = Card('9', 'h')
>>> hasattr(c4, 'new_attr')
False
"""
if __name__ == "__main__":
import doctest
doctest.testmod()8、前端控制器模式 Front_Controller
前端控制器模式（Front Controller Pattern）是用来提供一个集中的请求处理机制，所有的请求都将由一个单一的处理程序处理。
该处理程序可以做认证/授权/记录日志，或者跟踪请求，然后把请求传给相应的处理程序。
from __future__ import annotations
from typing import Any
class MobileView:
def show_index_page(self) -> None:
print("Displaying mobile index page")
class TabletView:
def show_index_page(self) -> None:
print("Displaying tablet index page")
class Dispatcher:
def __init__(self) -> None:
self.mobile_view = MobileView()
self.tablet_view = TabletView()
def dispatch(self, request: Request) -> None:
"""
This function is used to dispatch the request based on the type of device.
If it is a mobile, then mobile view will be called and if it is a tablet,
then tablet view will be called.
Otherwise, an error message will be printed saying that cannot dispatch the request.
"""
if request.type == Request.mobile_type:
self.mobile_view.show_index_page()
elif request.type == Request.tablet_type:
self.tablet_view.show_index_page()
else:
print("Cannot dispatch the request")
class RequestController:
"""front controller"""
def __init__(self) -> None:
self.dispatcher = Dispatcher()
def dispatch_request(self, request: Any) -> None:
"""
This function takes a request object and sends it to the dispatcher.
"""
if isinstance(request, Request):
self.dispatcher.dispatch(request)
else:
print("request must be a Request object")
class Request:
"""request"""
mobile_type = "mobile"
tablet_type = "tablet"
def __init__(self, request):
self.type = None
request = request.lower()
if request == self.mobile_type:
self.type = self.mobile_type
elif request == self.tablet_type:
self.type = self.tablet_type
def main():
"""
>>> front_controller = RequestController()
>>> front_controller.dispatch_request(Request('mobile'))
Displaying mobile index page
>>> front_controller.dispatch_request(Request('tablet'))
Displaying tablet index page
>>> front_controller.dispatch_request(Request('desktop'))
Cannot dispatch the request
>>> front_controller.dispatch_request('mobile')
request must be a Request object
"""
if __name__ == "__main__":
import doctest
doctest.testmod()9、MVC 模式
MVC 不仅仅是一种实现用户界面的软件模式，同时也是一种易于修改和维护的架构。通常 MVC 模式将应用程序分为 3 个基本部分：模型（Model）、视图（View）和控制器（Controller）。这 3 个部分相互关联，有助于将信息的处理与信息的呈现分开。
MVC 模式的工作机制为：模型提供数据和业务逻辑（如何存储和查询信息），视图负责数据的展示（如何呈现），而控制器则是两者之间的粘合剂，根据用户要求的呈现方式协调模型和视图。视图和控制器依赖于模型，但模型是可以独立工作的。
- 模型：定义针对数据的所有操作（如创建、修改和删除等），并提供与数据使用有关的方法
- 视图：提供相应的方法，帮助根据上下文和应用程序的需要构建 Web 或 GUI 界面
- 控制器：从请求接收数据，并将其发送到系统的其他部分。需要提供用于路由请求的方法
from abc import ABC, abstractmethod
class Model(ABC):
@abstractmethod
def __iter__(self):
pass
@abstractmethod
def get(self, item):
"""Returns an object with a .items() call method
that iterates over key,value pairs of its information."""
pass
@property
@abstractmethod
def item_type(self):
pass
class ProductModel(Model):
class Price(float):
"""A polymorphic way to pass a float with a particular
__str__ functionality."""
def __str__(self):
return f"{self:.2f}"
products = {
"milk": {"price": Price(1.50), "quantity": 10},
"eggs": {"price": Price(0.20), "quantity": 100},
"cheese": {"price": Price(2.00), "quantity": 10},
}
item_type = "product"
def __iter__(self):
yield from self.products
def get(self, product):
try:
return self.products[product]
except KeyError as e:
raise KeyError(str(e) + " not in the model's item list.")
class View(ABC):
@abstractmethod
def show_item_list(self, item_type, item_list):
pass
@abstractmethod
def show_item_information(self, item_type, item_name, item_info):
"""Will look for item information by iterating over key,value pairs
yielded by item_info.items()"""
pass
@abstractmethod
def item_not_found(self, item_type, item_name):
pass
class ConsoleView(View):
def show_item_list(self, item_type, item_list):
print(item_type.upper() + " LIST:")
for item in item_list:
print(item)
print("")
@staticmethod
def capitalizer(string):
return string[0].upper() + string[1:].lower()
def show_item_information(self, item_type, item_name, item_info):
print(item_type.upper() + " INFORMATION:")
printout = "Name: %s" % item_name
for key, value in item_info.items():
printout += ", " + self.capitalizer(str(key)) + ": " + str(value)
printout += "\n"
print(printout)
def item_not_found(self, item_type, item_name):
print(f'That {item_type} "{item_name}" does not exist in the records')
class Controller:
def __init__(self, model, view):
self.model = model
self.view = view
def show_items(self):
items = list(self.model)
item_type = self.model.item_type
self.view.show_item_list(item_type, items)
def show_item_information(self, item_name):
"""
Show information about a {item_type} item.
:param str item_name: the name of the {item_type} item to show information about
"""
try:
item_info = self.model.get(item_name)
except Exception:
item_type = self.model.item_type
self.view.item_not_found(item_type, item_name)
else:
item_type = self.model.item_type
self.view.show_item_information(item_type, item_name, item_info)
def main():
"""
>>> model = ProductModel()
>>> view = ConsoleView()
>>> controller = Controller(model, view)
>>> controller.show_items()
PRODUCT LIST:
milk
eggs
cheese
<BLANKLINE>
>>> controller.show_item_information("cheese")
PRODUCT INFORMATION:
Name: cheese, Price: 2.00, Quantity: 10
<BLANKLINE>
>>> controller.show_item_information("eggs")
PRODUCT INFORMATION:
Name: eggs, Price: 0.20, Quantity: 100
<BLANKLINE>
>>> controller.show_item_information("milk")
PRODUCT INFORMATION:
Name: milk, Price: 1.50, Quantity: 10
<BLANKLINE>
>>> controller.show_item_information("arepas")
That product "arepas" does not exist in the records
"""
if __name__ == "__main__":
import doctest
doctest.testmod()10、代理模式 Proxy
Python中的代理模式是一种设计模式，它提供了一种间接访问另一个对象的方式，以控制对原始对象的访问。代理模式通常用于需要对原始对象进行控制或保护时，或者需要向客户端隐藏原始对象的实现细节时。
例如 Python 里的引用计数。
from typing import Union
class Subject:
"""
As mentioned in the document, interfaces of both RealSubject and Proxy should
be the same, because the client should be able to use RealSubject or Proxy with
no code change.
Not all times this interface is necessary. The point is the client should be
able to use RealSubject or Proxy interchangeably with no change in code.
"""
def do_the_job(self, user: str) -> None:
raise NotImplementedError()
class RealSubject(Subject):
"""
This is the main job doer. External services like payment gateways can be a
good example.
"""
def do_the_job(self, user: str) -> None:
print(f"I am doing the job for {user}")
class Proxy(Subject):
def __init__(self) -> None:
self._real_subject = RealSubject()
def do_the_job(self, user: str) -> None:
"""
logging and controlling access are some examples of proxy usages.
"""
print(f"[log] Doing the job for {user} is requested.")
if user == "admin":
self._real_subject.do_the_job(user)
else:
print("[log] I can do the job just for `admins`.")
def client(job_doer: Union[RealSubject, Proxy], user: str) -> None:
job_doer.do_the_job(user)
def main():
"""
>>> proxy = Proxy()
>>> real_subject = RealSubject()
>>> client(proxy, 'admin')
[log] Doing the job for admin is requested.
I am doing the job for admin
>>> client(proxy, 'anonymous')
[log] Doing the job for anonymous is requested.
[log] I can do the job just for `admins`.
>>> client(real_subject, 'admin')
I am doing the job for admin
>>> client(real_subject, 'anonymous')
I am doing the job for anonymous
"""
if __name__ == "__main__":
import doctest
doctest.testmod()三、行为型模式
1、责任链模式 Chain_of_responsibility
责任链（Chain of Responsibility）模式的定义：为了避免请求发送者与多个请求处理者耦合在一起，于是将所有请求的处理者通过前一对象记住其下一个对象的引用而连成一条链；当有请求发生时，可将请求沿着这条链传递，直到有对象处理它为止。
注意：责任链模式也叫职责链模式。
在责任链模式中，客户只需要将请求发送到责任链上即可，无须关心请求的处理细节和请求的传递过程，请求会自动进行传递。所以责任链将请求的发送者和请求的处理者解耦了。
责任链模式是一种对象行为型模式，其主要优点如下：
- 降低了对象之间的耦合度。该模式使得一个对象无须知道到底是哪一个对象处理其请求以及链的结构，发送者和接收者也无须拥有对方的明确信息。
- 增强了系统的可扩展性。可以根据需要增加新的请求处理类，满足开闭原则。
- 增强了给对象指派职责的灵活性。当工作流程发生变化，可以动态地改变链内的成员或者调动它们的次序，也可动态地新增或者删除责任。
- 责任链简化了对象之间的连接。每个对象只需保持一个指向其后继者的引用，不需保持其他所有处理者的引用，这避免了使用众多的 if 或者 if···else 语句。
- 责任分担。每个类只需要处理自己该处理的工作，不该处理的传递给下一个对象完成，明确各类的责任范围，符合类的单一职责原则。
其主要缺点如下：
- 不能保证每个请求一定被处理。由于一个请求没有明确的接收者，所以不能保证它一定会被处理，该请求可能一直传到链的末端都得不到处理。
- 对比较长的职责链，请求的处理可能涉及多个处理对象，系统性能将受到一定影响。
- 职责链建立的合理性要靠客户端来保证，增加了客户端的复杂性，可能会由于职责链的错误设置而导致系统出错，如可能会造成循环调用。
from abc import ABC, abstractmethod
from typing import Optional, Tuple
class Handler(ABC):
def __init__(self, successor: Optional["Handler"] = None):
self.successor = successor
def handle(self, request: int) -> None:
"""
Handle request and stop.
If can't - call next handler in chain.
As an alternative you might even in case of success
call the next handler.
"""
res = self.check_range(request)
if not res and self.successor:
self.successor.handle(request)
@abstractmethod
def check_range(self, request: int) -> Optional[bool]:
"""Compare passed value to predefined interval"""
class ConcreteHandler0(Handler):
"""Each handler can be different.
Be simple and static...
"""
@staticmethod
def check_range(request: int) -> Optional[bool]:
if 0 <= request < 10:
print(f"request {request} handled in handler 0")
return True
return None
class ConcreteHandler1(Handler):
"""... With it's own internal state"""
start, end = 10, 20
def check_range(self, request: int) -> Optional[bool]:
if self.start <= request < self.end:
print(f"request {request} handled in handler 1")
return True
return None
class ConcreteHandler2(Handler):
"""... With helper methods."""
def check_range(self, request: int) -> Optional[bool]:
start, end = self.get_interval_from_db()
if start <= request < end:
print(f"request {request} handled in handler 2")
return True
return None
@staticmethod
def get_interval_from_db() -> Tuple[int, int]:
return (20, 30)
class FallbackHandler(Handler):
@staticmethod
def check_range(request: int) -> Optional[bool]:
print(f"end of chain, no handler for {request}")
return False
def main():
"""
>>> h0 = ConcreteHandler0()
>>> h1 = ConcreteHandler1()
>>> h2 = ConcreteHandler2(FallbackHandler())
>>> h0.successor = h1
>>> h1.successor = h2
>>> requests = [2, 5, 14, 22, 18, 3, 35, 27, 20]
>>> for request in requests:
... h0.handle(request)
request 2 handled in handler 0
request 5 handled in handler 0
request 14 handled in handler 1
request 22 handled in handler 2
request 18 handled in handler 1
request 3 handled in handler 0
end of chain, no handler for 35
request 27 handled in handler 2
request 20 handled in handler 2
"""
if __name__ == "__main__":
import doctest
doctest.testmod(optionflags=doctest.ELLIPSIS)2、目录模式 Catalog
目录设计模式的核心逻辑在于我们在一个类当中以方法的形式提供许多种功能，我们将这些功能以目录的形式存储在一个dict当中。我们在创建实例的时候通过不同的参数获取不同的功能，使用方在使用的时候不感知具体的参数。
class Catalog:
"""catalog of multiple static methods that are executed depending on an init
parameter
"""
def __init__(self, param: str) -> None:
# dictionary that will be used to determine which static method is
# to be executed but that will be also used to store possible param
# value
self._static_method_choices = {
"param_value_1": self._static_method_1,
"param_value_2": self._static_method_2,
}
# simple test to validate param value
if param in self._static_method_choices.keys():
self.param = param
else:
raise ValueError(f"Invalid Value for Param: {param}")
@staticmethod
def _static_method_1() -> None:
print("executed method 1!")
@staticmethod
def _static_method_2() -> None:
print("executed method 2!")
def main_method(self) -> None:
"""will execute either _static_method_1 or _static_method_2
depending on self.param value
"""
self._static_method_choices[self.param]()
# Alternative implementation for different levels of methods
class CatalogInstance:
"""catalog of multiple methods that are executed depending on an init
parameter
"""
def __init__(self, param: str) -> None:
self.x1 = "x1"
self.x2 = "x2"
# simple test to validate param value
if param in self._instance_method_choices:
self.param = param
else:
raise ValueError(f"Invalid Value for Param: {param}")
def _instance_method_1(self) -> None:
print(f"Value {self.x1}")
def _instance_method_2(self) -> None:
print(f"Value {self.x2}")
_instance_method_choices = {
"param_value_1": _instance_method_1,
"param_value_2": _instance_method_2,
}
def main_method(self) -> None:
"""will execute either _instance_method_1 or _instance_method_2
depending on self.param value
"""
self._instance_method_choices[self.param].__get__(self)() # type: ignore
# type ignore reason: https://github.com/python/mypy/issues/10206
class CatalogClass:
"""catalog of multiple class methods that are executed depending on an init
parameter
"""
x1 = "x1"
x2 = "x2"
def __init__(self, param: str) -> None:
# simple test to validate param value
if param in self._class_method_choices:
self.param = param
else:
raise ValueError(f"Invalid Value for Param: {param}")
@classmethod
def _class_method_1(cls) -> None:
print(f"Value {cls.x1}")
@classmethod
def _class_method_2(cls) -> None:
print(f"Value {cls.x2}")
_class_method_choices = {
"param_value_1": _class_method_1,
"param_value_2": _class_method_2,
}
def main_method(self):
"""will execute either _class_method_1 or _class_method_2
depending on self.param value
"""
self._class_method_choices[self.param].__get__(None, self.__class__)() # type: ignore
# type ignore reason: https://github.com/python/mypy/issues/10206
class CatalogStatic:
"""catalog of multiple static methods that are executed depending on an init
parameter
"""
def __init__(self, param: str) -> None:
# simple test to validate param value
if param in self._static_method_choices:
self.param = param
else:
raise ValueError(f"Invalid Value for Param: {param}")
@staticmethod
def _static_method_1() -> None:
print("executed method 1!")
@staticmethod
def _static_method_2() -> None:
print("executed method 2!")
_static_method_choices = {
"param_value_1": _static_method_1,
"param_value_2": _static_method_2,
}
def main_method(self) -> None:
"""will execute either _static_method_1 or _static_method_2
depending on self.param value
"""
self._static_method_choices[self.param].__get__(None, self.__class__)() # type: ignore
# type ignore reason: https://github.com/python/mypy/issues/10206
def main():
"""
>>> test = Catalog('param_value_2')
>>> test.main_method()
executed method 2!
>>> test = CatalogInstance('param_value_1')
>>> test.main_method()
Value x1
>>> test = CatalogClass('param_value_2')
>>> test.main_method()
Value x2
>>> test = CatalogStatic('param_value_1')
>>> test.main_method()
executed method 1!
"""
if __name__ == "__main__":
import doctest
doctest.testmod()3、方法链模式 Chaining_method
这个模式主要是将三个以上的method串起来，技巧体现在action类中的method可以串起来执行。
和chain模式异曲同工，都是讲一系列的动作串起来执行。
from __future__ import annotations
class Person:
def __init__(self, name: str) -> None:
self.name = name
def do_action(self, action: Action) -> Action:
print(self.name, action.name, end=" ")
return action
class Action:
def __init__(self, name: str) -> None:
self.name = name
def amount(self, val: str) -> Action:
print(val, end=" ")
return self
def stop(self) -> None:
print("then stop")
def main():
"""
>>> move = Action('move')
>>> person = Person('Jack')
>>> person.do_action(move).amount('5m').stop()
Jack move 5m then stop
"""
if __name__ == "__main__":
import doctest
doctest.testmod()4、命令模式 Command
一个类在进行工作时会调用自己或是其他类的方法，虽然调用结果会反映在对象的状态中，但并不会留下工作的历史记录。
这时，如果我们有一个类，用来表示“请进行这项工作”的“命令”就会方便很多。每一项想做的工作就不再是“方法的调用”这种动态处理了，而是一个表示命令的类的实例，即可以用“物”来表示。要想管理工作的历史记录，只需管理这些实例的集合即可，而且还可以随时再次执行过去的命令，或是将多个过去的命令整合为一个新命令并执行。
在设计模式中，我们称这样的“命令”为Command模式（command有“命令”的意思）。
Command有时也被称为事件（event）。它与“事件驱动编程”中的“事件”是一样的意思。当发生点击鼠标、按下键盘按键等事件时，我们可以先将这些事件作成实例，然后按照发生顺序放入队列中。接着，再依次去处理它们。在GUI（graphical user interface）编程中，经常需要与“事件”打交道。
from typing import List, Union
class HideFileCommand:
"""
A command to hide a file given its name
"""
def __init__(self) -> None:
# an array of files hidden, to undo them as needed
self._hidden_files: List[str] = []
def execute(self, filename: str) -> None:
print(f"hiding {filename}")
self._hidden_files.append(filename)
def undo(self) -> None:
filename = self._hidden_files.pop()
print(f"un-hiding {filename}")
class DeleteFileCommand:
"""
A command to delete a file given its name
"""
def __init__(self) -> None:
# an array of deleted files, to undo them as needed
self._deleted_files: List[str] = []
def execute(self, filename: str) -> None:
print(f"deleting {filename}")
self._deleted_files.append(filename)
def undo(self) -> None:
filename = self._deleted_files.pop()
print(f"restoring {filename}")
class MenuItem:
"""
The invoker class. Here it is items in a menu.
"""
def __init__(self, command: Union[HideFileCommand, DeleteFileCommand]) -> None:
self._command = command
def on_do_press(self, filename: str) -> None:
self._command.execute(filename)
def on_undo_press(self) -> None:
self._command.undo()
def main():
"""
>>> item1 = MenuItem(DeleteFileCommand())
>>> item2 = MenuItem(HideFileCommand())
# create a file named `test-file` to work with
>>> test_file_name = 'test-file'
# deleting `test-file`
>>> item1.on_do_press(test_file_name)
deleting test-file
# restoring `test-file`
>>> item1.on_undo_press()
restoring test-file
# hiding `test-file`
>>> item2.on_do_press(test_file_name)
hiding test-file
# un-hiding `test-file`
>>> item2.on_undo_press()
un-hiding test-file
"""
if __name__ == "__main__":
import doctest
doctest.testmod()5、迭代器模式 iterator
迭代器（Iterator）模式的定义：提供一个对象来顺序访问聚合对象中的一系列数据，而不暴露聚合对象的内部表示。迭代器模式是一种对象行为型模式，其主要优点如下。
- 访问一个聚合对象的内容而无须暴露它的内部表示。
- 遍历任务交由迭代器完成，这简化了聚合类。
- 它支持以不同方式遍历一个聚合，甚至可以自定义迭代器的子类以支持新的遍历。
- 增加新的聚合类和迭代器类都很方便，无须修改原有代码。
- 封装性良好，为遍历不同的聚合结构提供一个统一的接口。
其主要缺点是：增加了类的个数，这在一定程度上增加了系统的复杂性。
在日常开发中，我们几乎不会自己写迭代器。除非需要定制一个自己实现的数据结构对应的迭代器，否则，开源框架提供的 API 完全够用。
def count_to(count: int):
"""Counts by word numbers, up to a maximum of five"""
numbers = ["one", "two", "three", "four", "five"]
yield from numbers[:count]
# Test the generator
def count_to_two() -> None:
return count_to(2)
def count_to_five() -> None:
return count_to(5)
def main():
"""
# Counting to two...
>>> for number in count_to_two():
... print(number)
one
two
# Counting to five...
>>> for number in count_to_five():
... print(number)
one
two
three
four
five
"""
if __name__ == "__main__":
import doctest
doctest.testmod()from __future__ import annotations
class NumberWords:
"""Counts by word numbers, up to a maximum of five"""
_WORD_MAP = (
"one",
"two",
"three",
"four",
"five",
)
def __init__(self, start: int, stop: int) -> None:
self.start = start
self.stop = stop
def __iter__(self) -> NumberWords: # this makes the class an Iterable
return self
def __next__(self) -> str: # this makes the class an Iterator
if self.start > self.stop or self.start > len(self._WORD_MAP):
raise StopIteration
current = self.start
self.start += 1
return self._WORD_MAP[current - 1]
# Test the iterator
def main():
"""
# Counting to two...
>>> for number in NumberWords(start=1, stop=2):
... print(number)
one
two
# Counting to five...
>>> for number in NumberWords(start=1, stop=5):
... print(number)
one
two
three
four
five
"""
if __name__ == "__main__":
import doctest
doctest.testmod()6、中介模式 Mediator
中介者模式是一种行为型模式，是为了解决对象之间错综复杂的调用关系的一种设计模式，这种错综复杂的调用关系采用一个中介类来进行帮忙调用，所有的调用者只是需要关心中介者，而不需要进行互相依赖。
from __future__ import annotations
class ChatRoom:
"""Mediator class"""
def display_message(self, user: User, message: str) -> None:
print(f"[{user} says]: {message}")
class User:
"""A class whose instances want to interact with each other"""
def __init__(self, name: str) -> None:
self.name = name
self.chat_room = ChatRoom()
def say(self, message: str) -> None:
self.chat_room.display_message(self, message)
def __str__(self) -> str:
return self.name
def main():
"""
>>> molly = User('Molly')
>>> mark = User('Mark')
>>> ethan = User('Ethan')
>>> molly.say("Hi Team! Meeting at 3 PM today.")
[Molly says]: Hi Team! Meeting at 3 PM today.
>>> mark.say("Roger that!")
[Mark says]: Roger that!
>>> ethan.say("Alright.")
[Ethan says]: Alright.
"""
if __name__ == "__main__":
import doctest
doctest.testmod()7、备忘录模式 Memento
Memento(备忘录)模式：在不破坏封装性前提下，获取对象内部状态并外部保存，以方便日后恢复对象状态。
适用情况：
- 当需要保存一个对象的内部状态，以方便后续在用户出错时进行恢复。
- 如果一个接口让其它对象直接获取内部状态，又要不破坏对象的封装性。
from typing import Callable, List
from copy import copy, deepcopy
def memento(obj, deep=False):
state = deepcopy(obj.__dict__) if deep else copy(obj.__dict__)
def restore():
obj.__dict__.clear()
obj.__dict__.update(state)
return restore
class Transaction:
"""A transaction guard.
This is, in fact, just syntactic sugar around a memento closure.
"""
deep = False
states: List[Callable[[], None]] = []
def __init__(self, deep, *targets):
self.deep = deep
self.targets = targets
self.commit()
def commit(self):
self.states = [memento(target, self.deep) for target in self.targets]
def rollback(self):
for a_state in self.states:
a_state()
class Transactional:
"""Adds transactional semantics to methods. Methods decorated with
@Transactional will rollback to entry-state upon exceptions.
"""
def __init__(self, method):
self.method = method
def __get__(self, obj, T):
"""
A decorator that makes a function transactional.
:param method: The function to be decorated.
"""
def transaction(*args, **kwargs):
state = memento(obj)
try:
return self.method(obj, *args, **kwargs)
except Exception as e:
state()
raise e
return transaction
class NumObj:
def __init__(self, value):
self.value = value
def __repr__(self):
return f"<{self.__class__.__name__}: {self.value!r}>"
def increment(self):
self.value += 1
@Transactional
def do_stuff(self):
self.value = "1111" # <- invalid value
self.increment() # <- will fail and rollback
def main():
"""
>>> num_obj = NumObj(-1)
>>> print(num_obj)
<NumObj: -1>
>>> a_transaction = Transaction(True, num_obj)
>>> try:
... for i in range(3):
... num_obj.increment()
... print(num_obj)
... a_transaction.commit()
... print('-- committed')
... for i in range(3):
... num_obj.increment()
... print(num_obj)
... num_obj.value += 'x' # will fail
... print(num_obj)
... except Exception:
... a_transaction.rollback()
... print('-- rolled back')
<NumObj: 0>
<NumObj: 1>
<NumObj: 2>
-- committed
<NumObj: 3>
<NumObj: 4>
<NumObj: 5>
-- rolled back
>>> print(num_obj)
<NumObj: 2>
>>> print('-- now doing stuff ...')
-- now doing stuff ...
>>> try:
... num_obj.do_stuff()
... except Exception:
... print('-> doing stuff failed!')
... import sys
... import traceback
... traceback.print_exc(file=sys.stdout)
-> doing stuff failed!
Traceback (most recent call last):
...
TypeError: ...str...int...
>>> print(num_obj)
<NumObj: 2>
"""
if __name__ == "__main__":
import doctest
doctest.testmod(optionflags=doctest.ELLIPSIS)8、观察者模式 Observer
Observer(观察者)模式：定义对象间的一种一对多的依赖关系，当对象发生状态变化，依赖它的所有对象都将收到通知并自动改变。
适用性：
- 当一个抽象模型有两方面，一个依赖另一个，同时又希望它们独立变化，不紧耦合。
- 当一个对象改变影响多个对象，同时又不清楚具体多少对象受影响。
- 一个对象改变必须通知其它对象，同时不希望这些对象紧密耦合。
from __future__ import annotations
from contextlib import suppress
from typing import Protocol
# define a generic observer type
class Observer(Protocol):
def update(self, subject: Subject) -> None:
pass
class Subject:
def __init__(self) -> None:
self._observers: list[Observer] = []
def attach(self, observer: Observer) -> None:
if observer not in self._observers:
self._observers.append(observer)
def detach(self, observer: Observer) -> None:
with suppress(ValueError):
self._observers.remove(observer)
def notify(self, modifier: Observer | None = None) -> None:
for observer in self._observers:
if modifier != observer:
observer.update(self)
class Data(Subject):
def __init__(self, name: str = "") -> None:
super().__init__()
self.name = name
self._data = 0
@property
def data(self) -> int:
return self._data
@data.setter
def data(self, value: int) -> None:
self._data = value
self.notify()
class HexViewer:
def update(self, subject: Data) -> None:
print(f"HexViewer: Subject {subject.name} has data 0x{subject.data:x}")
class DecimalViewer:
def update(self, subject: Data) -> None:
print(f"DecimalViewer: Subject {subject.name} has data {subject.data}")
def main():
"""
>>> data1 = Data('Data 1')
>>> data2 = Data('Data 2')
>>> view1 = DecimalViewer()
>>> view2 = HexViewer()
>>> data1.attach(view1)
>>> data1.attach(view2)
>>> data2.attach(view2)
>>> data2.attach(view1)
>>> data1.data = 10
DecimalViewer: Subject Data 1 has data 10
HexViewer: Subject Data 1 has data 0xa
>>> data2.data = 15
HexViewer: Subject Data 2 has data 0xf
DecimalViewer: Subject Data 2 has data 15
>>> data1.data = 3
DecimalViewer: Subject Data 1 has data 3
HexViewer: Subject Data 1 has data 0x3
>>> data2.data = 5
HexViewer: Subject Data 2 has data 0x5
DecimalViewer: Subject Data 2 has data 5
# Detach HexViewer from data1 and data2
>>> data1.detach(view2)
>>> data2.detach(view2)
>>> data1.data = 10
DecimalViewer: Subject Data 1 has data 10
>>> data2.data = 15
DecimalViewer: Subject Data 2 has data 15
"""
if __name__ == "__main__":
import doctest
doctest.testmod()9、发布订阅模式 Publish/Subscribe
Publish/Subscribe模式比较观察者模式则多了一个类似于话题调度中心的流程，发布者和订阅者解耦。
观察者模式更像是去咖啡店点咖啡，向店家点一杯咖啡，然后做好后店家会送过来，我和店家是直接有交互的。
发布订阅模式就像是我们微信里面订阅公众号，发布者把文章发布到这个公众号，我们订阅公众号后就会收到发布者发布的文章，我们可以关注和取消关注这个公众号。
from __future__ import annotations
class Provider:
def __init__(self) -> None:
self.msg_queue = []
self.subscribers = {}
def notify(self, msg: str) -> None:
self.msg_queue.append(msg)
def subscribe(self, msg: str, subscriber: Subscriber) -> None:
self.subscribers.setdefault(msg, []).append(subscriber)
def unsubscribe(self, msg: str, subscriber: Subscriber) -> None:
self.subscribers[msg].remove(subscriber)
def update(self) -> None:
for msg in self.msg_queue:
for sub in self.subscribers.get(msg, []):
sub.run(msg)
self.msg_queue = []
class Publisher:
def __init__(self, msg_center: Provider) -> None:
self.provider = msg_center
def publish(self, msg: str) -> None:
self.provider.notify(msg)
class Subscriber:
def __init__(self, name: str, msg_center: Provider) -> None:
self.name = name
self.provider = msg_center
def subscribe(self, msg: str) -> None:
self.provider.subscribe(msg, self)
def unsubscribe(self, msg: str) -> None:
self.provider.unsubscribe(msg, self)
def run(self, msg: str) -> None:
print(f"{self.name} got {msg}")
def main():
"""
>>> message_center = Provider()
>>> fftv = Publisher(message_center)
>>> jim = Subscriber("jim", message_center)
>>> jim.subscribe("cartoon")
>>> jack = Subscriber("jack", message_center)
>>> jack.subscribe("music")
>>> gee = Subscriber("gee", message_center)
>>> gee.subscribe("movie")
>>> vani = Subscriber("vani", message_center)
>>> vani.subscribe("movie")
>>> vani.unsubscribe("movie")
# Note that no one subscribed to `ads`
# and that vani changed their mind
>>> fftv.publish("cartoon")
>>> fftv.publish("music")
>>> fftv.publish("ads")
>>> fftv.publish("movie")
>>> fftv.publish("cartoon")
>>> fftv.publish("cartoon")
>>> fftv.publish("movie")
>>> fftv.publish("blank")
>>> message_center.update()
jim got cartoon
jack got music
gee got movie
jim got cartoon
jim got cartoon
gee got movie
"""
if __name__ == "__main__":
import doctest
doctest.testmod()10、注册模式 Registry
注册模式（Registry）也叫做注册树模式，注册器模式。注册模式为应用中经常使用的对象创建一个中央存储器来存放这些对象 —— 通常通过一个只包含静态方法的抽象类来实现（或者通过单例模式）。
from typing import Dict
class RegistryHolder(type):
REGISTRY: Dict[str, "RegistryHolder"] = {}
def __new__(cls, name, bases, attrs):
new_cls = type.__new__(cls, name, bases, attrs)
"""
Here the name of the class is used as key but it could be any class
parameter.
"""
cls.REGISTRY[new_cls.__name__] = new_cls
return new_cls
@classmethod
def get_registry(cls):
return dict(cls.REGISTRY)
class BaseRegisteredClass(metaclass=RegistryHolder):
"""
Any class that will inherits from BaseRegisteredClass will be included
inside the dict RegistryHolder.REGISTRY, the key being the name of the
class and the associated value, the class itself.
"""
def main():
"""
Before subclassing
>>> sorted(RegistryHolder.REGISTRY)
['BaseRegisteredClass']
>>> class ClassRegistree(BaseRegisteredClass):
... def __init__(self, *args, **kwargs):
... pass
After subclassing
>>> sorted(RegistryHolder.REGISTRY)
['BaseRegisteredClass', 'ClassRegistree']
"""
if __name__ == "__main__":
import doctest
doctest.testmod(optionflags=doctest.ELLIPSIS)11、规格模式 Specification
规格模式（Specification Pattern）可以认为是组合模式的一种扩展。很多时候程序中的某些条件决定了业务逻辑，这些条件就可以抽离出来以某种关系（与、或、非）进行组合，从而灵活地对业务逻辑进行定制。
另外，在查询、过滤等应用场合中，通过预定义多个条件，然后使用这些条件的组合来处理查询或过滤，而不是使用逻辑判断语句来处理，可以简化整个实现逻辑。这里的每个条件都是一个规格，多个规格（条件）通过串联的方式以某种逻辑关系形成一个组合式的规格。规格模式属于结构型设计模式。
from abc import abstractmethod
class Specification:
def and_specification(self, candidate):
raise NotImplementedError()
def or_specification(self, candidate):
raise NotImplementedError()
def not_specification(self):
raise NotImplementedError()
@abstractmethod
def is_satisfied_by(self, candidate):
pass
class CompositeSpecification(Specification):
@abstractmethod
def is_satisfied_by(self, candidate):
pass
def and_specification(self, candidate):
return AndSpecification(self, candidate)
def or_specification(self, candidate):
return OrSpecification(self, candidate)
def not_specification(self):
return NotSpecification(self)
class AndSpecification(CompositeSpecification):
def __init__(self, one, other):
self._one: Specification = one
self._other: Specification = other
def is_satisfied_by(self, candidate):
return bool(
self._one.is_satisfied_by(candidate)
and self._other.is_satisfied_by(candidate)
)
class OrSpecification(CompositeSpecification):
def __init__(self, one, other):
self._one: Specification = one
self._other: Specification = other
def is_satisfied_by(self, candidate):
return bool(
self._one.is_satisfied_by(candidate)
or self._other.is_satisfied_by(candidate)
)
class NotSpecification(CompositeSpecification):
def __init__(self, wrapped):
self._wrapped: Specification = wrapped
def is_satisfied_by(self, candidate):
return bool(not self._wrapped.is_satisfied_by(candidate))
class User:
def __init__(self, super_user=False):
self.super_user = super_user
class UserSpecification(CompositeSpecification):
def is_satisfied_by(self, candidate):
return isinstance(candidate, User)
class SuperUserSpecification(CompositeSpecification):
def is_satisfied_by(self, candidate):
return getattr(candidate, "super_user", False)
def main():
"""
>>> andrey = User()
>>> ivan = User(super_user=True)
>>> vasiliy = 'not User instance'
>>> root_specification = UserSpecification().and_specification(SuperUserSpecification())
# Is specification satisfied by <name>
>>> root_specification.is_satisfied_by(andrey), 'andrey'
(False, 'andrey')
>>> root_specification.is_satisfied_by(ivan), 'ivan'
(True, 'ivan')
>>> root_specification.is_satisfied_by(vasiliy), 'vasiliy'
(False, 'vasiliy')
"""
if __name__ == "__main__":
import doctest
doctest.testmod()12、状态模式 State
在面向对象编程中，是用类表示对象的。也就是说，程序的设计者需要考虑用类来表示什么东西。类对应的东西可能存在于真实世界中，也可能不存在于真实世界中。
在State模式中，我们用类来表示状态。在现实世界中，我们会考虑各种东西的“状态”，但是几乎不会将状态当作“东西”看待。因此，可能大家很难理解“用类来表示状态”的意思。
用一句话来概括：State模式就是用类来表示状态。
from __future__ import annotations
class State:
"""Base state. This is to share functionality"""
def scan(self) -> None:
"""Scan the dial to the next station"""
self.pos += 1
if self.pos == len(self.stations):
self.pos = 0
print(f"Scanning... Station is {self.stations[self.pos]} {self.name}")
class AmState(State):
def __init__(self, radio: Radio) -> None:
self.radio = radio
self.stations = ["1250", "1380", "1510"]
self.pos = 0
self.name = "AM"
def toggle_amfm(self) -> None:
print("Switching to FM")
self.radio.state = self.radio.fmstate
class FmState(State):
def __init__(self, radio: Radio) -> None:
self.radio = radio
self.stations = ["81.3", "89.1", "103.9"]
self.pos = 0
self.name = "FM"
def toggle_amfm(self) -> None:
print("Switching to AM")
self.radio.state = self.radio.amstate
class Radio:
"""A radio. It has a scan button, and an AM/FM toggle switch."""
def __init__(self) -> None:
"""We have an AM state and an FM state"""
self.amstate = AmState(self)
self.fmstate = FmState(self)
self.state = self.amstate
def toggle_amfm(self) -> None:
self.state.toggle_amfm()
def scan(self) -> None:
self.state.scan()
def main():
"""
>>> radio = Radio()
>>> actions = [radio.scan] * 2 + [radio.toggle_amfm] + [radio.scan] * 2
>>> actions *= 2
>>> for action in actions:
... action()
Scanning... Station is 1380 AM
Scanning... Station is 1510 AM
Switching to FM
Scanning... Station is 89.1 FM
Scanning... Station is 103.9 FM
Scanning... Station is 81.3 FM
Scanning... Station is 89.1 FM
Switching to AM
Scanning... Station is 1250 AM
Scanning... Station is 1380 AM
"""
if __name__ == "__main__":
import doctest
doctest.testmod()13、策略模式 Strategy
策略模式就是定义一系列的算法，把它们一个个封装起来，并且使它们可以相互替换。这个模式可以使得算法独立于使用它的客户变化而变化。
有许多算法可以对数组进行排序，将这些排序算法编写进使用它们的类中是不取的。原因是因为如果将排序算法编写进使用的类中将会使系统变得复杂，并且难以维护。不同的时候可能需要使用不同的排序算法，并不需要支持不使用的排序算法。当排序算法是程序中必不可少的一部分时，增加新的排序算法或者改变现有的算法将会变得非常困难。我们可以定义一些类来封装这些算法，即一个一个策略。
适用性：许多相关的类仅仅是行为有异；需要使用一个算法的不同变体；算法使用客户应该不知道的数据；一个类中定义了多种行为，并且这些行为在这个类的操作中以多条件语句的形式出现。
from __future__ import annotations
from typing import Callable
class DiscountStrategyValidator: # Descriptor class for check perform
@staticmethod
def validate(obj: Order, value: Callable) -> bool:
try:
if obj.price - value(obj) < 0:
raise ValueError(
f"Discount cannot be applied due to negative price resulting. {value.__name__}"
)
except ValueError as ex:
print(str(ex))
return False
else:
return True
def __set_name__(self, owner, name: str) -> None:
self.private_name = f"_{name}"
def __set__(self, obj: Order, value: Callable = None) -> None:
if value and self.validate(obj, value):
setattr(obj, self.private_name, value)
else:
setattr(obj, self.private_name, None)
def __get__(self, obj: object, objtype: type = None):
return getattr(obj, self.private_name)
class Order:
discount_strategy = DiscountStrategyValidator()
def __init__(self, price: float, discount_strategy: Callable = None) -> None:
self.price: float = price
self.discount_strategy = discount_strategy
def apply_discount(self) -> float:
if self.discount_strategy:
discount = self.discount_strategy(self)
else:
discount = 0
return self.price - discount
def __repr__(self) -> str:
return f"<Order price: {self.price} with discount strategy: {getattr(self.discount_strategy,'__name__',None)}>"
def ten_percent_discount(order: Order) -> float:
return order.price * 0.10
def on_sale_discount(order: Order) -> float:
return order.price * 0.25 + 20
def main():
"""
>>> order = Order(100, discount_strategy=ten_percent_discount)
>>> print(order)
<Order price: 100 with discount strategy: ten_percent_discount>
>>> print(order.apply_discount())
90.0
>>> order = Order(100, discount_strategy=on_sale_discount)
>>> print(order)
<Order price: 100 with discount strategy: on_sale_discount>
>>> print(order.apply_discount())
55.0
>>> order = Order(10, discount_strategy=on_sale_discount)
Discount cannot be applied due to negative price resulting. on_sale_discount
>>> print(order)
<Order price: 10 with discount strategy: None>
"""
if __name__ == "__main__":
import doctest
doctest.testmod()14、模板模式 Template
做一件是的方法很多，但做这件都可以归纳为几个步骤。这个时候可以使用模板模式，在模板类中，定义做事的步骤，将多种实现做事的细节延迟到子类中去实现。
即：定义一个操作中的算法的骨架（模板函数），而将一些步骤延迟到子类中（基本函数）。模板方法使得子类可以不改变一个算法的结构（模板函数）即可重定义该算法的实现方式（基本函数）。
开闭原则：对修改关闭、对扩展开发
依赖倒置（DIP dependency inversion principle）：高层次模块不依赖与低层次模块（调用模块不依赖于被调用模块的修改和扩展，如果被调用模块是基于抽象类开发，那么调用模块只要基于抽象类调用即可。这样就出现了依赖倒置，即具体的类要依赖于抽象的类来实现，而不是抽象类或者调用模块依赖于被调用模块）。依赖倒置的原则就是：子类依赖于抽象基类实现，调用模块依赖于抽象基类调用，这样隔离了调用模块与具体实现的子类的依赖。（要求抽象基类定义的很好）
def get_text() -> str:
return "plain-text"
def get_pdf() -> str:
return "pdf"
def get_csv() -> str:
return "csv"
def convert_to_text(data: str) -> str:
print("[CONVERT]")
return f"{data} as text"
def saver() -> None:
print("[SAVE]")
def template_function(getter, converter=False, to_save=False) -> None:
data = getter()
print(f"Got `{data}`")
if len(data) <= 3 and converter:
data = converter(data)
else:
print("Skip conversion")
if to_save:
saver()
print(f"`{data}` was processed")
def main():
"""
>>> template_function(get_text, to_save=True)
Got `plain-text`
Skip conversion
[SAVE]
`plain-text` was processed
>>> template_function(get_pdf, converter=convert_to_text)
Got `pdf`
[CONVERT]
`pdf as text` was processed
>>> template_function(get_csv, to_save=True)
Got `csv`
Skip conversion
[SAVE]
`csv` was processed
"""
if __name__ == "__main__":
import doctest
doctest.testmod()15、访问者模式 Visitor
访问者（Visitor）模式的定义：将作用于某种数据结构中的各元素的操作分离出来封装成独立的类，使其在不改变数据结构的前提下可以添加作用于这些元素的新的操作，为数据结构中的每个元素提供多种访问方式。它将对数据的操作与数据结构进行分离，是行为类模式中最复杂的一种模式。
访问者（Visitor）模式是一种对象行为型模式，其主要优点如下：
- 扩展性好。能够在不修改对象结构中的元素的情况下，为对象结构中的元素添加新的功能。
- 复用性好。可以通过访问者来定义整个对象结构通用的功能，从而提高系统的复用程度。
- 灵活性好。访问者模式将数据结构与作用于结构上的操作解耦，使得操作集合可相对自由地演化而不影响系统的数据结构。
- 符合单一职责原则。访问者模式把相关的行为封装在一起，构成一个访问者，使每一个访问者的功能都比较单一。
访问者（Visitor）模式的主要缺点如下：
- 增加新的元素类很困难。在访问者模式中，每增加一个新的元素类，都要在每一个具体访问者类中增加相应的具体操作，这违背了“开闭原则”。
- 破坏封装。访问者模式中具体元素对访问者公布细节，这破坏了对象的封装性。
- 违反了依赖倒置原则。访问者模式依赖了具体类，而没有依赖抽象类。
class Node:
pass
class A(Node):
pass
class B(Node):
pass
class C(A, B):
pass
class Visitor:
def visit(self, node, *args, **kwargs):
meth = None
for cls in node.__class__.__mro__:
meth_name = "visit_" + cls.__name__
meth = getattr(self, meth_name, None)
if meth:
break
if not meth:
meth = self.generic_visit
return meth(node, *args, **kwargs)
def generic_visit(self, node, *args, **kwargs):
print("generic_visit " + node.__class__.__name__)
def visit_B(self, node, *args, **kwargs):
print("visit_B " + node.__class__.__name__)
def main():
"""
>>> a, b, c = A(), B(), C()
>>> visitor = Visitor()
>>> visitor.visit(a)
generic_visit A
>>> visitor.visit(b)
visit_B B
>>> visitor.visit(c)
visit_B C
"""
if __name__ == "__main__":
import doctest
doctest.testmod()四、其他模型
1、依赖注入模式 Dependency Injection
依赖注入是一种软件设计模式，其中一个或多个依赖项（或服务）被注入或通过引用传递到依赖对象（或客户端），并成为客户端状态的一部分。该模式将客户端依赖项的创建与其自身行为分开，从而允许程序设计松散耦合，并遵循控制反转和单一责任原则。
适用场景
- 当你需要从对象中删除具体实现的知识时；
- 使用模拟对象或存根来隔离类的单元测试；
import datetime
from typing import Callable
class ConstructorInjection:
def __init__(self, time_provider: Callable) -> None:
self.time_provider = time_provider
def get_current_time_as_html_fragment(self) -> str:
current_time = self.time_provider()
current_time_as_html_fragment = '<span class="tinyBoldText">{}</span>'.format(
current_time
)
return current_time_as_html_fragment
class ParameterInjection:
def __init__(self) -> None:
pass
def get_current_time_as_html_fragment(self, time_provider: Callable) -> str:
current_time = time_provider()
current_time_as_html_fragment = '<span class="tinyBoldText">{}</span>'.format(
current_time
)
return current_time_as_html_fragment
class SetterInjection:
"""Setter Injection"""
def __init__(self):
pass
def set_time_provider(self, time_provider: Callable):
self.time_provider = time_provider
def get_current_time_as_html_fragment(self):
current_time = self.time_provider()
current_time_as_html_fragment = '<span class="tinyBoldText">{}</span>'.format(
current_time
)
return current_time_as_html_fragment
def production_code_time_provider() -> str:
"""
Production code version of the time provider (just a wrapper for formatting
datetime for this example).
"""
current_time = datetime.datetime.now()
current_time_formatted = f"{current_time.hour}:{current_time.minute}"
return current_time_formatted
def midnight_time_provider() -> str:
"""Hard-coded stub"""
return "24:01"
def main():
"""
>>> time_with_ci1 = ConstructorInjection(midnight_time_provider)
>>> time_with_ci1.get_current_time_as_html_fragment()
'<span class="tinyBoldText">24:01</span>'
>>> time_with_ci2 = ConstructorInjection(production_code_time_provider)
>>> time_with_ci2.get_current_time_as_html_fragment()
'<span class="tinyBoldText">...</span>'
>>> time_with_pi = ParameterInjection()
>>> time_with_pi.get_current_time_as_html_fragment(midnight_time_provider)
'<span class="tinyBoldText">24:01</span>'
>>> time_with_si = SetterInjection()
>>> time_with_si.get_current_time_as_html_fragment()
Traceback (most recent call last):
...
AttributeError: 'SetterInjection' object has no attribute 'time_provider'
>>> time_with_si.set_time_provider(midnight_time_provider)
>>> time_with_si.get_current_time_as_html_fragment()
'<span class="tinyBoldText">24:01</span>'
"""
if __name__ == "__main__":
import doctest
doctest.testmod(optionflags=doctest.ELLIPSIS)2、委托模式 Delegation_pattern
委托者对象提供原本的功能，被委托者对象提供附加功能。
有两个对象参与(两个对象协同)处理同一个请求，接受请求的对象将请求委托给另一个对象来处理。
from __future__ import annotations
from typing import Any, Callable
class Delegator:
"""
>>> delegator = Delegator(Delegate())
>>> delegator.p1
123
>>> delegator.p2
Traceback (most recent call last):
...
AttributeError: 'Delegate' object has no attribute 'p2'
>>> delegator.do_something("nothing")
'Doing nothing'
>>> delegator.do_anything()
Traceback (most recent call last):
...
AttributeError: 'Delegate' object has no attribute 'do_anything'
"""
def __init__(self, delegate: Delegate) -> None:
self.delegate = delegate
def __getattr__(self, name: str) -> Any | Callable:
attr = getattr(self.delegate, name)
if not callable(attr):
return attr
def wrapper(*args, **kwargs):
return attr(*args, **kwargs)
return wrapper
class Delegate:
def __init__(self) -> None:
self.p1 = 123
def do_something(self, something: str) -> str:
return f"Doing {something}"
if __name__ == "__main__":
import doctest
doctest.testmod()3、黑板模式 Blackboard
板模式是观察者模式的一个扩展，知名度并不高，但是我们使用的范围却非常广。
黑板模式的意图：允许消息的读写同时进行，广泛的交互消息。
简单的说，黑板模式允许多个消息读写者同时存在，消息的生产者和消费者完全分开。这就像一个黑板，任何一个教授（消息的生产者）都可以在其上书写消息，任何一个学生（消息的消费者）都可以从黑板上读取消息，两者在空间和时间上可以解耦，并且互不干扰。
黑板模式确实是消息的广播，主要解决的问题是消息的生产者和消费者之间的耦合问题，他的核心是消息存储（黑板），他存储所有消息，并可以随时被读取。当消息生产者把消息写入到消息仓库后，其他消息者就可以从仓库中读取。当然，此时消息的写入者也可以变身为消息的阅读者，读写者在时间上解耦。对于这些消息，消费者只需要关注特定消息，不处理与自己不相关的消息，这一点通常通过过滤器来实现。
from __future__ import annotations
import abc
import random
class Blackboard:
def __init__(self) -> None:
self.experts = []
self.common_state = {
"problems": 0,
"suggestions": 0,
"contributions": [],
"progress": 0, # percentage, if 100 -> task is finished
}
def add_expert(self, expert: AbstractExpert) -> None:
self.experts.append(expert)
class Controller:
def __init__(self, blackboard: Blackboard) -> None:
self.blackboard = blackboard
def run_loop(self):
"""
This function is a loop that runs until the progress reaches 100.
It checks if an expert is eager to contribute and then calls its contribute method.
"""
while self.blackboard.common_state["progress"] < 100:
for expert in self.blackboard.experts:
if expert.is_eager_to_contribute:
expert.contribute()
return self.blackboard.common_state["contributions"]
class AbstractExpert(metaclass=abc.ABCMeta):
def __init__(self, blackboard: Blackboard) -> None:
self.blackboard = blackboard
@property
@abc.abstractmethod
def is_eager_to_contribute(self):
raise NotImplementedError("Must provide implementation in subclass.")
@abc.abstractmethod
def contribute(self):
raise NotImplementedError("Must provide implementation in subclass.")
class Student(AbstractExpert):
@property
def is_eager_to_contribute(self) -> bool:
return True
def contribute(self) -> None:
self.blackboard.common_state["problems"] += random.randint(1, 10)
self.blackboard.common_state["suggestions"] += random.randint(1, 10)
self.blackboard.common_state["contributions"] += [self.__class__.__name__]
self.blackboard.common_state["progress"] += random.randint(1, 2)
class Scientist(AbstractExpert):
@property
def is_eager_to_contribute(self) -> int:
return random.randint(0, 1)
def contribute(self) -> None:
self.blackboard.common_state["problems"] += random.randint(10, 20)
self.blackboard.common_state["suggestions"] += random.randint(10, 20)
self.blackboard.common_state["contributions"] += [self.__class__.__name__]
self.blackboard.common_state["progress"] += random.randint(10, 30)
class Professor(AbstractExpert):
@property
def is_eager_to_contribute(self) -> bool:
return True if self.blackboard.common_state["problems"] > 100 else False
def contribute(self) -> None:
self.blackboard.common_state["problems"] += random.randint(1, 2)
self.blackboard.common_state["suggestions"] += random.randint(10, 20)
self.blackboard.common_state["contributions"] += [self.__class__.__name__]
self.blackboard.common_state["progress"] += random.randint(10, 100)
def main():
"""
>>> blackboard = Blackboard()
>>> blackboard.add_expert(Student(blackboard))
>>> blackboard.add_expert(Scientist(blackboard))
>>> blackboard.add_expert(Professor(blackboard))
>>> c = Controller(blackboard)
>>> contributions = c.run_loop()
>>> from pprint import pprint
>>> pprint(contributions)
['Student',
'Student',
'Student',
'Student',
'Scientist',
'Student',
'Student',
'Student',
'Scientist',
'Student',
'Scientist',
'Student',
'Student',
'Scientist',
'Professor']
"""
if __name__ == "__main__":
random.seed(1234) # for deterministic doctest outputs
import doctest
doctest.testmod()4、图搜索模式 Graph_search
class GraphSearch:
"""Graph search emulation in python, from source
http://www.python.org/doc/essays/graphs/
dfs stands for Depth First Search
bfs stands for Breadth First Search"""
def __init__(self, graph):
self.graph = graph
def find_path_dfs(self, start, end, path=None):
path = path or []
path.append(start)
if start == end:
return path
for node in self.graph.get(start, []):
if node not in path:
newpath = self.find_path_dfs(node, end, path[:])
if newpath:
return newpath
def find_all_paths_dfs(self, start, end, path=None):
path = path or []
path.append(start)
if start == end:
return [path]
paths = []
for node in self.graph.get(start, []):
if node not in path:
newpaths = self.find_all_paths_dfs(node, end, path[:])
paths.extend(newpaths)
return paths
def find_shortest_path_dfs(self, start, end, path=None):
path = path or []
path.append(start)
if start == end:
return path
shortest = None
for node in self.graph.get(start, []):
if node not in path:
newpath = self.find_shortest_path_dfs(node, end, path[:])
if newpath:
if not shortest or len(newpath) < len(shortest):
shortest = newpath
return shortest
def find_shortest_path_bfs(self, start, end):
"""
Finds the shortest path between two nodes in a graph using breadth-first search.
:param start: The node to start from.
:type start: str or int
:param end: The node to find the shortest path to.
:type end: str or int
:returns queue_path_to_end, dist_to[end]: A list of nodes
representing the shortest path from `start` to `end`, and a dictionary
mapping each node in the graph (except for `start`) with its distance from it
(in terms of hops). If no such path exists, returns an empty list and an empty
dictionary instead.
"""
queue = [start]
dist_to = {start: 0}
edge_to = {}
if start == end:
return queue
while len(queue):
value = queue.pop(0)
for node in self.graph[value]:
if node not in dist_to.keys():
edge_to[node] = value
dist_to[node] = dist_to[value] + 1
queue.append(node)
if end in edge_to.keys():
path = []
node = end
while dist_to[node] != 0:
path.insert(0, node)
node = edge_to[node]
path.insert(0, start)
return path
def main():
"""
# example of graph usage
>>> graph = {
... 'A': ['B', 'C'],
... 'B': ['C', 'D'],
... 'C': ['D', 'G'],
... 'D': ['C'],
... 'E': ['F'],
... 'F': ['C'],
... 'G': ['E'],
... 'H': ['C']
... }
# initialization of new graph search object
>>> graph_search = GraphSearch(graph)
>>> print(graph_search.find_path_dfs('A', 'D'))
['A', 'B', 'C', 'D']
# start the search somewhere in the middle
>>> print(graph_search.find_path_dfs('G', 'F'))
['G', 'E', 'F']
# unreachable node
>>> print(graph_search.find_path_dfs('C', 'H'))
None
# non existing node
>>> print(graph_search.find_path_dfs('C', 'X'))
None
>>> print(graph_search.find_all_paths_dfs('A', 'D'))
[['A', 'B', 'C', 'D'], ['A', 'B', 'D'], ['A', 'C', 'D']]
>>> print(graph_search.find_shortest_path_dfs('A', 'D'))
['A', 'B', 'D']
>>> print(graph_search.find_shortest_path_dfs('A', 'F'))
['A', 'C', 'G', 'E', 'F']
>>> print(graph_search.find_shortest_path_bfs('A', 'D'))
['A', 'B', 'D']
>>> print(graph_search.find_shortest_path_bfs('A', 'F'))
['A', 'C', 'G', 'E', 'F']
# start the search somewhere in the middle
>>> print(graph_search.find_shortest_path_bfs('G', 'F'))
['G', 'E', 'F']
# unreachable node
>>> print(graph_search.find_shortest_path_bfs('A', 'H'))
None
# non existing node
>>> print(graph_search.find_shortest_path_bfs('A', 'X'))
None
"""
if __name__ == "__main__":
import doctest
doctest.testmod()5、Hsm模式 Hsm
class UnsupportedMessageType(BaseException):
pass
class UnsupportedState(BaseException):
pass
class UnsupportedTransition(BaseException):
pass
class HierachicalStateMachine:
def __init__(self):
self._active_state = Active(self) # Unit.Inservice.Active()
self._standby_state = Standby(self) # Unit.Inservice.Standby()
self._suspect_state = Suspect(self) # Unit.OutOfService.Suspect()
self._failed_state = Failed(self) # Unit.OutOfService.Failed()
self._current_state = self._standby_state
self.states = {
"active": self._active_state,
"standby": self._standby_state,
"suspect": self._suspect_state,
"failed": self._failed_state,
}
self.message_types = {
"fault trigger": self._current_state.on_fault_trigger,
"switchover": self._current_state.on_switchover,
"diagnostics passed": self._current_state.on_diagnostics_passed,
"diagnostics failed": self._current_state.on_diagnostics_failed,
"operator inservice": self._current_state.on_operator_inservice,
}
def _next_state(self, state):
try:
self._current_state = self.states[state]
except KeyError:
raise UnsupportedState
def _send_diagnostics_request(self):
return "send diagnostic request"
def _raise_alarm(self):
return "raise alarm"
def _clear_alarm(self):
return "clear alarm"
def _perform_switchover(self):
return "perform switchover"
def _send_switchover_response(self):
return "send switchover response"
def _send_operator_inservice_response(self):
return "send operator inservice response"
def _send_diagnostics_failure_report(self):
return "send diagnostics failure report"
def _send_diagnostics_pass_report(self):
return "send diagnostics pass report"
def _abort_diagnostics(self):
return "abort diagnostics"
def _check_mate_status(self):
return "check mate status"
def on_message(self, message_type): # message ignored
if message_type in self.message_types.keys():
self.message_types[message_type]()
else:
raise UnsupportedMessageType
class Unit:
def __init__(self, HierachicalStateMachine):
self.hsm = HierachicalStateMachine
def on_switchover(self):
raise UnsupportedTransition
def on_fault_trigger(self):
raise UnsupportedTransition
def on_diagnostics_failed(self):
raise UnsupportedTransition
def on_diagnostics_passed(self):
raise UnsupportedTransition
def on_operator_inservice(self):
raise UnsupportedTransition
class Inservice(Unit):
def __init__(self, HierachicalStateMachine):
self._hsm = HierachicalStateMachine
def on_fault_trigger(self):
self._hsm._next_state("suspect")
self._hsm._send_diagnostics_request()
self._hsm._raise_alarm()
def on_switchover(self):
self._hsm._perform_switchover()
self._hsm._check_mate_status()
self._hsm._send_switchover_response()
class Active(Inservice):
def __init__(self, HierachicalStateMachine):
self._hsm = HierachicalStateMachine
def on_fault_trigger(self):
super().perform_switchover()
super().on_fault_trigger()
def on_switchover(self):
self._hsm.on_switchover() # message ignored
self._hsm.next_state("standby")
class Standby(Inservice):
def __init__(self, HierachicalStateMachine):
self._hsm = HierachicalStateMachine
def on_switchover(self):
super().on_switchover() # message ignored
self._hsm._next_state("active")
class OutOfService(Unit):
def __init__(self, HierachicalStateMachine):
self._hsm = HierachicalStateMachine
def on_operator_inservice(self):
self._hsm.on_switchover() # message ignored
self._hsm.send_operator_inservice_response()
self._hsm.next_state("suspect")
class Suspect(OutOfService):
def __init__(self, HierachicalStateMachine):
self._hsm = HierachicalStateMachine
def on_diagnostics_failed(self):
super().send_diagnostics_failure_report()
super().next_state("failed")
def on_diagnostics_passed(self):
super().send_diagnostics_pass_report()
super().clear_alarm() # loss of redundancy alarm
super().next_state("standby")
def on_operator_inservice(self):
super().abort_diagnostics()
super().on_operator_inservice() # message ignored
class Failed(OutOfService):
"""No need to override any method."""
def __init__(self, HierachicalStateMachine):
self._hsm = HierachicalStateMachine