Python Design Patterns
如何写出更好的代码——刘欣 2018.6.20
"Design patterns help you learn from others’ successes instead of your own failures"
断断续续读了好久设计模式，记录此篇总结。设计模式最初发源于C++/Java等静态语言，Python语言本身的很多特性已经覆盖了设计模式，甚至用了都不知道，比如：decorator/metaclass/generator/getattr等。但是写稍微大型的项目时，还是经常力不从心。就如上面引用的那句话，通过设计模式可以学习前人的智慧，写出更好的代码。
所有代码都在https://github.com/Meteorix/python-design-patterns，python3环境下可以跑通，请跑起来玩玩。代码仅限演示作用，更注重清晰地用python语法展示patterns，而不是完备性，请勿用在生产环境。欢迎提issue和pr : )
设计模式六大原则
-
单一职责原则
- 一个类只做一件事情，模块化
-
里氏替换原则
- 所有使用父类的地方必须能完全替换为使用其子类
- 即：子类可以扩展父类的功能，但不能改变父类原有的功能
-
依赖倒置原则
- 高层模块不应该依赖低层模块，二者都应该依赖其抽象
- 抽象不应该依赖实现；实现应该依赖抽象
- 面向接口编程，而不是面向实现编程，Duck Type
-
接口隔离原则
- 一个类对另一个类依赖的接口越少越好
-
最小知识原则
- 一个类对另一个类知道得越少越好
-
开闭原则
- 类、模块、函数对扩展开放，对修改关闭
- 尽量在不修改源代码的情况下进行扩展
其实以上的原则不限于类的设计，很多工程上的系统设计也适用。
常用设计模式
创造模式
Singleton
一个类只有一个对象，似乎不太需要解释：）
class SingletonMeta(type):
instance = None
def __call__(cls, *args, **kwargs):
if cls.instance is None:
cls.instance = super(SingletonMeta, cls).__call__(*args, **kwargs)
return cls.instance
class CurrentUser(object, metaclass=SingletonMeta):
def __init__(self, name=None):
super(CurrentUser, self).__init__()
self.name = name
def __str__(self):
return repr(self) + ":" + repr(self.name)
if __name__ == '__main__':
u = CurrentUser("liu")
print(u)
u2 = CurrentUser()
u2.name = "xin"
print(u2)
print(u)
assert u is u2这个例子用MetaClass实现，其实Python里还有其他实现方式。但是Python里的MetaClass就是用来实例化Class的，用来实现单例类正好。
Factory
工厂模式，用于生产一大堆对象。这里用__subclasses__来获取子类，这样可以动态扩展子类而不改变factory的代码。
class Shape(object):
@classmethod
def factory(cls, name, *args, **kwargs):
types = {c.__name__: c for c in cls.__subclasses__()} # 忽略性能:P
shape_class = types[name]
return shape_class(*args, **kwargs)
class Circle(Shape):
pass
class Square(Shape):
pass
if __name__ == '__main__':
shapes = ["Circle", "Square", "Square", "Circle"]
for i in shapes:
s = Shape.factory(i)
print(s)结构模式
MVC
可能是最有名的设计模式，数据<->控制器<->视图。数据和视图分离，还可以同一份数据渲染多个视图。MVC做的最好的应该是各种Web框架和GUI框架。
class Model(object):
products = {
'milk': {'price': 1.50, 'quantity': 10},
'eggs': {'price': 0.20, 'quantity': 100},
'cheese': {'price': 2.00, 'quantity': 10}
}
def get(self, name):
return self.products.get(name)
class View(object):
def show_item_list(self, item_list):
print('-' * 20)
for item in item_list:
print("* Name: %s" % item)
print('-' * 20)
def show_item_info(self, name, item_info):
print("Name: %s Price: %s Quantity: %s" % (name, item_info['price'], item_info['quantity']))
print('-' * 20)
def show_empty(self, name):
print("Name: %s not found" % name)
print('-' * 20)
class Controller(object):
def __init__(self, model, view):
self.model = model
self.view = view
def show_items(self):
items = self.model.products.keys()
self.view.show_item_list(items)
def show_item_info(self, item):
item_info = self.model.get(item)
if item_info:
self.view.show_item_info(item, item_info)
else:
self.view.show_empty(item)
if __name__ == '__main__':
model = Model()
view = View()
controller = Controller(model, view)
controller.show_items()
controller.show_item_info('cheese')
controller.show_item_info('apple')上面的例子还只演示了数据到视图的渲染，其实MVC还包括通过视图修改数据。
Proxy
不直接调用一个类，而是通过一个代理来访问。这样做的好处有：可以切换底层实现、权限控制、安全检查等。当然最有用的是可以实现远程代理，jsonrpc就是一种。
class Implementation(object):
def add(self, x, y):
return x + y
def minus(self, x, y):
return x - y
class Proxy(object):
def __init__(self, impl):
self._impl = impl
def __getattr__(self, name):
return getattr(self._impl, name)
if __name__ == '__main__':
p = Proxy(Implementation())
print(p.add(1, 2))
print(p.minus(1, 2))Decorator
装饰器，似乎不太需要解释，Python自带的语法，可以用来做很多事情，几个简单例子：
- 路由
from flask import Flask
app = Flask(__name__)
@app.route('/')
def index():
return 'Hello, World'
@app.route('/home')
def home():
return 'Welcome Home'- 权限控制
from django.contrib.auth.decorators import login_required
@login_required
def my_view(request):
...- 输入输出
from functools import wraps
def debug(f):
@wraps(f)
def debug_function(*args, **kwargs):
print('call: ', f.__name__, args, kwargs)
ret = f(*args, **kwargs)
print('return: ', ret)
return debug_function
@debug
def foo(a, b, c=None):
print(a, b, c)
return True
if __name__ == '__main__':
foo(1, 2, 3)行为模式
Template
基类作为模板，定义好接口，子类来实现功能，最好的例子就是Qt里的各种QWidget。
class ApplicateFramework(object):
def __init__(self):
self.setup()
self.show()
def setup(self):
pass
def show(self):
pass
def close(self):
pass
class MyApplication(ApplicateFramework):
def setup(self):
print("setup", self)
def show(self):
print("show", self)
def close(self):
print("close", self)
if __name__ == '__main__':
app = MyApplication()
app.close()State Machine
状态机，当前状态 + 操作 => 下一个状态，似乎也不用怎么解释。如下例子实现的状态机，可以自定义状态、操作和转换规则。扩展的时候无需修改状态机代码，符合开闭原则。
class StateMachine(object):
def __init__(self, init_state):
self.current_state = init_state
self.current_state.run()
def step(self, action):
self.current_state = self.current_state.next(action)
self.current_state.run()
class State(object):
def __init__(self, name):
self.name = name
def __str__(self):
return "<State '%s'>" % self.name
def next(self, action):
if (self, action) in mapping:
next_state = mapping[(self, action)]
else:
next_state = self
print("%s + %s => %s" % (self, action, next_state))
return next_state
def run(self):
print(self, "is current state")
class Action(object):
def __init__(self, name):
self.name = name
def __str__(self):
return "<Action '%s'>" % self.name
State.Running = State("Running")
State.Stopped = State("Stopped")
State.Paused = State("Paused")
Action.start = Action("start")
Action.stop = Action("stop")
Action.pause = Action("pause")
Action.resume = Action("resume")
mapping = {
(State.Stopped, Action.start): State.Running,
(State.Running, Action.stop): State.Stopped,
(State.Running, Action.pause): State.Paused,
(State.Paused, Action.resume): State.Running,
(State.Paused, Action.stop): State.Stopped,
}
if __name__ == '__main__':
state_machine = StateMachine(State.Stopped)
state_machine.step(Action.start)
state_machine.step(Action.pause)
state_machine.step(Action.resume)
state_machine.step(Action.stop)Iterator
迭代器，在Python中也不需要怎么解释，使用起来好像理所应当一样。实际上迭代器的一大优势是无需关心数据类型，一样的for语法。另一大优势是无需事先计算好所有元素，而是在迭代到的时候才计算。如下例子是Python中使用yield语法产生生成器generator来实现的迭代器，优势一目了然。
def fibonacci(count=100):
a, b = 1, 2
yield a
yield b
while count:
a, b = b, a + b
count -= 1
yield b
for i in fibonacci():
print(i)Command
Command封装了一个原子操作，在类外面实现，个人认为最大的作用是实现redo/undo。
from collections import deque
class Document(object):
value = ""
cmd_stack = deque()
@classmethod
def execute(cls, cmd):
cmd.redo()
cls.cmd_stack.append(cmd)
@classmethod
def undo(cls):
cmd = cls.cmd_stack.pop()
cmd.undo()
class AddTextCommand(object):
def __init__(self, text):
self.text = text
def redo(self):
Document.value += self.text
def undo(self):
Document.value = Document.value[:-len(self.text)]
if __name__ == '__main__':
cmds = [AddTextCommand("liu"), AddTextCommand("xin"), AddTextCommand("heihei")]
for cmd in cmds:
Document.execute(cmd)
print(Document.value)
for i in range(len(cmds)):
Document.undo()
print(Document.value)Chain Of Responsibility
链式Handler处理请求，某一个处理成功就返回。用Chain来动态构造Handler序列。
class Handler(object):
def __init__(self):
self.successor = None
def handle(self, data):
res = self._handle(data)
if res:
return res
if self.successor:
return self.successor.handle(data)
def _handle(self, data):
raise NotImplementedError
def link(self, handler):
self.successor = handler
return handler
class DictHandler(Handler):
def _handle(self, data):
if isinstance(data, dict):
print("handled by %s" % self)
return True
class ListHandler(Handler):
def _handle(self, data):
if isinstance(data, list):
print("handled by %s" % self)
return True
if __name__ == '__main__':
h = DictHandler()
h.link(ListHandler()).link(Handler())
ret = h.handle([1, 2, 3])
ret = h.handle({1: 2})Chaining Method
链式反应，就这样一直点下去。很多Query构造函数是这样，API更好用。
class Player(object):
def __init__(self, name):
self.pos = (0, 0)
def move(self, pos):
self.pos = pos
print("move to %s, %s" % self.pos)
return self
def say(self, text):
print(text)
return self
def home(self):
self.pos = (0, 0)
print("I am home")
return self
if __name__ == '__main__':
p = Player('liuxin')
p.move((1, 1)).say("haha").move((2, 3)).home().say("go to sleep")Visitor
Visitor模式的目的是不改变原来的类，用另一个类来实现一些接口。下面的例子用Visitor模式实现了两种节点遍历的方法。
class Node(object):
def __init__(self, name, children=()):
self.name = name
self.children = list(children)
def __str__(self):
return '<Node %s>' % self.name
class Visitor(object):
@classmethod
def visit(cls, node):
yield node
for child in node.children:
yield child
@classmethod
def visit2(cls, node):
for child in node.children:
yield child
yield node
if __name__ == '__main__':
root = Node('root', (Node('a'), Node('b')))
visitor = Visitor()
for node in visitor.visit(root):
print(node)
for node in visitor.visit2(root):
print(node)Observer
当一个对象发生状态变化时，需要更新其他对象，用观察者模式来解耦这些对象，最小知识原则。
class Observable(object):
def __init__(self):
self._observers = []
def attach(self, observer):
if observer not in self._observers:
self._observers.append(observer)
def detach(self, observer):
self._observers.remove(observer)
def notify(self):
for observer in self._observers:
observer.update(self)
class Observer(object):
def update(self, observable):
print('updating %s by %s' % (self, observable))
if __name__ == '__main__':
clock = Observable()
user1 = Observer()
user2 = Observer()
clock.attach(user1)
clock.attach(user2)
clock.notify()上面的例子演示了最简单的实现，通常在实际程序中观察者模式会在不同线程中，要注意线程安全的问题。