Python设计模式笔记 - Pursue wisdom love fun

Creatial Patterns

Abstract Factory

抽象工厂模式：提供一个接口创建对象，但不需要指定他们真实的类，适用于创建抽象对象依赖不同的配置、平台选择等

class PetShop:
    def __init__(self, animal_factory=None):
        self.pet_factory = animal_factory

    def show_pet(self):
        pet = self.pet_factory()
        print("We have a lovely {}".format(pet))
        print("It says {}".format(pet.speak()))

class Dog:
    def speak(self):
        return "woof"

    def __str__(self):
        return "Dog"

class Cat:
    def speak(self):
        return "maomao"

    def __str__(self):
        return "Cat"

if __name__ == "__main__":
    cat_shop = PetShop(cat)
    cat_shop.show_pet()

Factory

工厂模式：一个函数创建其他的对象

class DevConfig:
    DEBUG = False

class ProdConfig:
    DEBUG = True

def get_debug(config='dev'):
    config = {
        "dev": DevConfig,
        "prod": ProdConfig
    }

    return config[config].DEBUG


if __name__ == "__main__":
    get_debug('dev')

Borg

Borg: 多个实例之间分享相同的状态

class Borg:
    __shared_state = {} # 类中的私有变量

    def __init__(self):
        self.__dict__ = self.__shared_state
        self.state = "Init"

    def __str__(self):
        return self.state

class YourBorg(Borg):
    pass

if __name__ == "__main__":
    b1 = Borg()
    b2 = Borg()
    b1.state = 'hhh'
    b2.state = 'wcg'
    print(b1, b2) # wcg wcg

Builder

构造模式：解耦一个复杂对象的创建和表示，通常抽象

class Building:
    def __init__(self):
        self.build_floor()
        self.build_size()

    def build_floor(self):
        raise NotImplementedError

    def build_size(self):
        raise NotImplementedError

    def __repr__(self):
        return 'Floor: {0.floor} | Size: {0.size}'.format(self)

class House(Building):
    def build_floor(self):
        self.floor = 'One'

    def build_size(self):
        self.size = 'Big'

class Flat(Building):
    def build_floor(self):
        self.floor = 'More than One'

    def build_size(self):
        self.size = 'Smail'

if __name__ == "__main__":
    house = House()
    flat = Flat()

Lazy_evaluation

惰性求值：初始化的时候不计算，调用的时候才计算

import functools

class lazy_property:
    def __init__(self, func):
        self.func = func
        functools.update_wrapper(self, func)

    def __get__(self, obj, type_):
        if obj is None:
            return self
        val = self.func(obj)
        obj.__dict__[self.func.__name__] = val
        return val

def lazy_property2(fn):
    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_count = 0

    @lazy_property
    def relatives(self):
        relatives = "Many relatives"
        return relatives

    @lazy_property2
    def parents(self):
        self.call_count += 1
        return "Father and mother"

if __name__ == "__main__":
    Jhon = Person('Jhon', 'Coder')
    Json.name # Jhon
    Json.occupation # Coder
    Json.__dict__.items() # [('name': 'Json', 'occupation': 'Coder', 'call_count': 0)]
    Json.relatives # 'Many relatives'
    Json.__dict__.items() # [('relatives': 'Many relatives')]
    Json.parents # 'Father and mother'
    Json.__dict__.items() # [('_lazy__parents': 'Father and mother' ...)]
    Json.parents # 'Father and mother'
    Json.call_count # 1 call_count不会变

Pool

池：保存相同的一组实例

class ObjectPool(object):
    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

if __name__ == "__main__":
    import queue

    sample_queue = queue.Queue()
    sample_queue.put('yam')
    with ObjectPool(sample_queue) as obj:
        print('Inside with: {}'.format(obj))
    print('Outside with: {}'.format(sample_queue.get()))

    # Inside with: yam
    # Outside with: yam

Prototype

原型：通过克隆原型创建实例，减少类

class Prototype:
    value = 'dafault'

    def clone(self, **attrs):
        obj = self.__class__
        obj.__dict__.update(attrs)
        return obj

class PrototypeDispatcher:
    def __init__(self):
        self._objects = {}

    def get_objects(self):
        return self._objects

    def register_object(self, name, obj):
        self._objects[name] = obj

    def unregister_object(self, name):
        del self._objects[name]

def main():
    dispatcher = PrototypeDispatcher()
    prototype = Prototype()

    d = prototype.clone()
    a = prototype.clone(value='a-value', category='a')
    b = prototype.clone(value='b-value', is_checked=True)
    dispatcher.register_object('default', d)
    dispatcher.register_object('objecta', a)
    dispatcher.register_object('objectb', b)
    print([{n: p.value} for n, p in dispatcher.get_objects().items()])
    # [{'default': 'default'}, {'objecta': 'a-value'}, {'objectb': 'b-value'}]

Structural Patterns

3-tier

3-tier: 分离演示，应用程序处理和数据管理,一层层封装

class Data:
    products = {
        'milk': {'price': 1.50, 'quantity': 10},
        'eggs': {'price': 0.20, 'quantity': 100},
        'cheese': {'price': 2.00, 'quantity': 10},
    }

    def __get__(self, obj, klas):
        return {'products': self.products}

class BussinessLogic:
    data = Data()

    def product_list(self):
        return self.data['products'].keys()

    def product_information(self, product):
        return self.data['products'].get(product, None)

class UI:
    def __init__(self):
        self.business_logic = BussinessLogic()

    def get_product_list(self):
        for product in self.business_logic.product_list():
            print(product)

    def get_prouduct_informatin(self, product):
        pass

def main():
    ui = UI()
    ui.get_product_list()
    ui.get_prouduct_informatin('eggs')

adapter

适配器：通过引入间接层来实现不兼容接口的适配，可以用继承，也可以用dict属性

class Dog:
    def __init__(self):
        self.name = 'dog'

    def bark(self):
        return 'wangwang'

class Cat:
    def __init__(self):
        self.name = 'cat'

    def meow(self):
        return 'maomao'

class Car:
    def __init__(self):
        self.name = name

    def make_noise(self):
        return 'dididi'

class Adapter:
    def __init__(self, obj, **adapted_methods):
        self.obj = obj
        self.__dict__.update(adapted_methods)

    def __getattr__(self, attr): # 从obj获取
        return getattr(self.obj, attr)

    def original_dict(self):
        return self.obj.__dict__

def main():
    dog = Dog()
    dog.__dict__ # {'name': 'Dog'}
    a = Adapter(dog, make_noise=dog.bark)
    a.__dict__ # {'obj': <__main__.Dog at xxx>, 'make_noise': <bound method Dog.bark of <__main__.Dog at xxx>}
    a.bark == a.make_noise # True
    a.make_noise() # 'wangwang'

bridge

桥模式：将抽象和实现分离

class DrawingApi1:
    def draw_circle(self, x, y, radius):
        print('Api1.circle at {}:{} radius {}'.format(x, y, radius))

class DrawingApi2:
    def draw_circle(self, x, y, radius):
        print('Api2.circle at {}:{} radius {}'.format(x, y, radius))

class CircleShape:
    def __init__(self, x, y, radius, drawing_api):
        self._x = x
        self._y = y
        self._radius = radius
        self._drawing_api = drawing_api

    def draw(self):
        self._drawing_api.draw_circle(self._x, self._y, self._radius)

    def scale(self, pct):
        self._radius *= pct

def main():
    shape = CircleShape(1, 2, 3, DrawingApi1())
    shape.draw()

composite

综合模式：让客户端统一处理单个对象和组合

class Graphic:
    def render(self):
        raise NotImplementedError("You should implement this")

class CompositeGraphic(Graphic):
    def __init__(self):
        self.graphics = []

    def render(self):
        for graphic in self.graphics:
            graphic.render()

    def add(self, graphic):
        self.graphics.append(graphic)

    def remove(self, graphic):
        self.graphics.remove(graphic)

class Ellipse(Graphic):
    def __init__(self, name):
        self.name = name

    def render(self):
        print("Ellipse: {}".format(self.name))

def main():
    pass

decorator

装饰器模式：不改变实例，动态给对象增加新的功能

class TextTag:
    def __init__(self, text):
        self._text = text

    def render(self):
        return self._text

class BoldWrapper(TextTag):
    def __init__(self, wrapped):
        self._wrapped = wrapped

    def redner(self):
        return '<b>{}</b>'.format(self._wrapped.render())

class ItalicWrapper(TextTag):
    def __init__(self, wrapped):
        self._wrapped = wrapped

    def render(self):
        return '<i>{}</i>'.format(self._wrapped.render())

if __name__ == '__main__':
    simple_hello = TextTag('hello, world')
    special_hello = ItalicWrapper(BoldWrapper(simple_hello))
    simple_hello.render()  # hello, world
    special_hello.render() # <i><b>hello, world</b></i>

facade

facade: 提供一个简单统一的接口来隐藏复杂的系统

class CPU:
    def freeze(self):
        pass
    def jump(self):
        pass
    def execute(self):
        pass

class Memory:
    def load(self):
        pass

class Computer:
    def __init__(self):
        self.cpu = CPU()
        self.memory = Memory()

    def start(self):
        self.cpu.freeze()
        self.memory.load()
        self.cpu.jump()
        self.cpu.execute()

flyweight

flyweight: 最小化对象的数量,对象池

import weakref

class Card:
    _pool = weakref.WeakValueDictionary() # 弱引用, 1 解决交叉引用 2 剩下的引用是弱映射对象所持有的弱引用时,会垃圾回收

    def __new__(cls, value, suit):
        obj = cls._pool.get(value + suit)
        if obj is None:
            obj = object.__new__(Card)
            cls._pool[value + suit] = obj
            obj.value, obj.suit = value, suit
        return obj

    def __repr__(self):
        return "<Card: %s%s>" % (self.value, self.suit)

front_controller

front_controller: 提供一个集中的入口点来管理和控制请求处理

# 类似django
class MobileView:
    def get(self):
        print('')

class TableView:
    def get(self):
        print('')

class Dispatcher:
    def __init__(self):
        self.moblie_view = MobileView()
        self.tablet_view = TableView()

    def dispatch(self, request):
        if request.type == Request.mobile_type:
            self.moblie_view.get()
        elif reqest.type == request.tablet_type:
            self.tablet_view.get()
        else:
            print('')

class RequestController:
    def __init__(self):
        self.dispatcher = Dispatcher()

    def dispatch_request(self, request):
        if isinstance(request, Requeset):
            self.dispatcher.dispatch(request)
        else:
            print('')

class 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

mvc

mvc: model-view-controller

class Model(object):
    def __iter__(self):
        raise NotImplementedError

    def get(self, item):
        """Returns an object with a .items() call method
        that iterates over key,value pairs of its information."""
        raise NotImplementedError

    @property
    def item_type(self):
        raise NotImplementedError


class ProductModel(Model):
    class Price(float):
        """A polymorphic way to pass a float with a particular
        __str__ functionality."""

        def __str__(self):
            return "{:.2f}".format(self)

    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):
        for item in self.products:
            yield item

    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(object):
    def show_item_list(self, item_type, item_list):
        raise NotImplementedError

    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()"""
        raise NotImplementedError

    def item_not_found(self, item_type, item_name):
        raise NotImplementedError


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('That %s "%s" does not exist in the records' % (item_type, item_name))


class Controller(object):
    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):
        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)

proxy

proxy: 提供复制资源的接口

from __future__ import print_function
import time


class SalesManager:
    def talk(self):
        print("Sales Manager ready to talk")


class Proxy:
    def __init__(self):
        self.busy = 'No'
        self.sales = None

    def talk(self):
        print("Proxy checking for Sales Manager availability")
        if self.busy == 'No':
            self.sales = SalesManager()
            time.sleep(0.1)
            self.sales.talk()
        else:
            time.sleep(0.1)
            print("Sales Manager is busy")


class NoTalkProxy(Proxy):
    def talk(self):
        print("Proxy checking for Sales Manager availability")
        time.sleep(0.1)
        print("This Sales Manager will not talk to you", "whether he/she is busy or not")

Behavioral Patterns

chain_of_responsibility

chain_of_responsibility : 在请求的发送者和请求解耦

import abc

class Handler(metaclass=abc.ABCMeta):
    def __init__(self, successor=None):
        self.successor = successor

    def handle(self, request):
        res = self.check_range(request)
        if not res and self.successor:
            self.successor.handle(request)

    @abc.abstractclassmethod
    def check_range(self, request):
        pass

class ConcreteHandler0(Handler):
    @staticmethod
    def check_range(request):
        if 0 <= request < 10:
            return True

class ConcreteHandler1(Handler):
    start, end = 10, 20

    def check_range(self, request):
        if self.start <= request <= self.end:
            return True

class ConcreteHandler2(Handler):
    def check_range(self, request):
        start, end = self.get_interval_from_db()
        if start <= request < end:
            return True

    @staticmethod
    def get_interval_from_db():
        return (20, 30)


class FallbackHandler(Handler):
    @staticmethod
    def check_range(request):
        return False

catalog

catalog: 根据传入的参数使用不同的静态方法

class Catalog(object):
    """catalog of multiple static methods that are executed depending on an init
    parameter
    """

    def __init__(self, param):

        # 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("Invalid Value for Param: {0}".format(param))

    @staticmethod
    def _static_method_1():
        print("executed method 1!")

    @staticmethod
    def _static_method_2():
        print("executed method 2!")

    def main_method(self):
        """will execute either _static_method_1 or _static_method_2
        depending on self.param value
        """
        self._static_method_choices[self.param]()

chaining_method

chaining_method: 继续回调下一个方法

class Person:
    def __init__(self, name, action):
        self.name = name
        self.action = action

    def do_action(self):
        return self.action

class Action:
    def __init__(self, name):
        self.name = name

    def amount(self, val):
        return self

    def stop(self):
        print('then stop')

def main():
    move = Action('move')
    person = Person('Jack', move)
    person.do_action().amount('5m').stop()

command

command: 绑定一个命令和参数来调用

class MoveFileCommand:
    def __init__(self, src, dest):
        self.src = src
        self.dest = dest

    def execute(self):
        self.rename(self.src, self.dest)

    def undo(self):
        self.rename(self.dest, self.src)

    def rename(self, src, dest):
        os.rename(src, dest)

iterator

iterator: 迭代器

def count_to(count):
    numbers = ['one', 'two', 'three', 'four', 'five']
    for number in numbers[:count]:
        yield number

count_to_two = lambda: count_to(2)
count_to_five = lambda: count_to(5)

def main():
    for number in count_to_two():
        print(number)

mediator

mediator: 中间人

class ChatRoom:
    def display_message(self, user, message):
        print("[{} says]: {}".format(user, message))

class User:
    def __init__(self, name):
        self.name = name
        self.chat_room = ChatRoom()

    def say(self, message):
        self.chat_room.display_message(self, message)

    def __str__(self):
        return self.name

def main():
    molly = User('Molly')
    molly.say("Hi Wcg")
    """[molly says]: Hi Wcg"""

memento

memento: 纪念, 生成不透明的令牌,可用于返回先前的状态

observer

observer: 观察者模式,提供回调以通知事件/数据更改

class Subject:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        if observer not in self._observers:
            self._observers.append(observer)

    def detach(self, observer):
        try:
            self._observers.remove(observer)
        except ValueError:
            pass

    def notify(self, modifier=None):
        for observer in self._observers:
            if modifier != observer:
                observer.update(self)

class Data(Subject):
    def __init__(self, name=''):
        Subject.__init__(self)
        self.name = name
        self._data = 0

    @property
    def data(self):
        return self._data

    @data.setattr
    def data(self, value):
        self._data = value
        self.notify()

class HexViewer:
    def update(self, subject):
        print('HexViewer: Subject %s has data 0x%x' % (subject.name, subject.data))

class DecimalViewer:
    def update(self, subject):
        print('DecimalViewer: Subject %s has data %d' % (subject.name, subject.data))

iterator: 发布/订阅

class Provider:
    def __init__(self):
        self.msg_queue = []
        self.subscribes = {}

    def notify(self, msg):
        self.msg_queue.append(msg)

    def subscribe(self, msg, subscriber):
        self.subscribes.setdefault(msg, []).append(subscriber)

    def unsubscribe(self, msg, subscriber):
        self.subscribes[msg].remove(subscriber)

    def update(self):
        for msg in self.msg_queue:
            for sub in self.subscribes.get(msg, []):
                sub.run(msg)
        self.msg_queue = []

class Publisher:
    """发布"""
    def __init__(self, msg_center):
        self.provider = msg_center

    def publish(self, msg):
        self.provider.notify(msg)

class Subscriber:
    """订阅"""
    def __init__(self, name, msg_center):
        self.name = name
        self.provider = msg_center

    def subscribe(self, msg):
        self.provider.subscribe(msg, self)

    def unsubscribe(self, msg):
        self.provider.unsubscribe(msg, self)

    def run(self, msg):
        print("{} got {}".format(self.name, 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")

    fftv.publish("cartoon")
    fftv.publish("music")

    message_center.update()

"""jim got cartoon
   jack got music"""

iterator

iterator: 迭代器

参考

https://github.com/faif/python-patterns
https://python-web-guide.readthedocs.io/zh/latest/design/design.html
https://github.com/itswcg/Books/blob/master/Head%20First%20%E8%AE%BE%E8%AE%A1%E6%A8%A1%E5%BC%8F.pdf>

Creatial Patterns

Abstract Factory

Factory

Borg

Builder

Lazy_evaluation

Pool

Prototype

Structural Patterns

3-tier

adapter

bridge

composite

decorator

facade

flyweight

front_controller

mvc

proxy

Behavioral Patterns

chain_of_responsibility

catalog

chaining_method

command

iterator

mediator

memento

observer

publish_subscribe

iterator

参考