
from abc import ABCMeta, abstractmethod
class Observable:
"""发布者基类"""
def __init__(self):
# 存储订阅者对象的列表
self.__observers = []
# 增加订阅者:必须方法,observer为订阅者类对象
def add_observer(self, observer):
self.__observers.append(observer)
# 删除订阅者:必须方法,observer为订阅者类对象
def remove_observer(self, observer):
self.__observers.remove(observer)
# 通知订阅者:同时调用订阅者类对象的方法update
# 推信息:发布者向订阅者传递参数,通过msg传给obj的update
# 拉信息:发布者将自身self,作为对象传给订阅者update
def notify_observers(self, msg=0):
for obj in self.__observers:
obj.update(self, msg)
class Account(Observable):
"""发布者具体实现:用户账户"""
def __init__(self):
super().__init__()
self.name = "用户账户"
# 此处调用了发布者基类的通知方法notifyObservers
def login(self, name, ip):
# 如果登陆验证不在白名单里,就出发通知方法
if not self.__is_auth(ip):
# 类方法默认第一个参数是self
self.notify_observers({"name": name, "ip": ip})
def __is_auth(self, ip):
# ip白名单
ip_white_list = [
"101.47.18.9"
]
# 如果在白名单里返回true
return ip in ip_white_list
class Observer(metaclass=ABCMeta):
"""订阅者基类"""
# 虚函数:此类的派生类必须重写此函数
# 拉信息:由订阅者主动从发布者类对象observable中提取
# 推信息(常用):由订阅者被动的从msg中拿数据
@abstractmethod
def update(self, observable, msg):
pass
class SmsSender(Observer):
"""订阅者实现:短信发送器"""
def update(self, observable, msg):
print("推模型:[短信发送] " + msg["name"] + "异常登陆IP:" + msg["ip"])
print("拉模型:拿到发布者的属性为", observable.name)
class MailSender(Observer):
"""订阅者实现:邮件发送器"""
def update(self, observable, msg):
print("[邮件发送] " + msg["name"] + "异常登陆IP:" + msg["ip"])
if __name__ == "__main__":
# 创建发布者
account = Account()
# 向发布者添加订阅者对象:短信
account.add_observer(SmsSender())
# 向发布者添加订阅者对象:电子邮件
account.add_observer(MailSender())
# 发布者状态发生变化
# 在白名单里
account.login("Sunny", "101.47.18.9")
# 不在白名单里
account.login("Tony", "67.218.147.69")
###################################################
>>> 推模型:[短信发送] Tony异常登陆IP:67.218.147.69
>>> 拉模型:拿到发布者的属性为 用户账户
>>> [邮件发送] Tony异常登陆IP:67.218.147.69


from abc import ABCMeta, abstractmethod
class Handler(metaclass=ABCMeta):
@abstractmethod
def handler_leave(self, day):
pass
# 责任链的终点:以下都是责任链上的节点
class GeneralManager(Handler):
def handler_leave(self, day):
if day < 10:
print(f"总经理:批准假{day}天")
else:
print("总经理:不批")
# 扩展:以此类为模板,可以扩展
class DepartmentMananger(Handler):
def __init__(self):
self.next = GeneralManager()
def handler_leave(self, day):
if day <= 5:
print(f"部门经理:批准假{day}天")
else:
# 此语句时建立责任链的纽带,数据结构的链式思维
print("部门经理:权限不足")
self.next.handler_leave(day)
class ProjectDirector(Handler):
def __init__(self):
self.next = DepartmentMananger()
def handler_leave(self, day):
if day <= 3:
print(f"项目主管:批准假{day}天")
else:
# 此语句时建立责任链的纽带,数据结构的链式思维
print("项目主管:权限不足")
self.next.handler_leave(day)
if __name__ == "__main__":
# 客户端
h = ProjectDirector()
print("请假2天时")
h.handler_leave(2)
print("请假15天时")
h.handler_leave(15)
#####################################################
>>> 请假2天时
>>> 项目主管:批准假2天
>>> 请假15天时
>>> 项目主管:权限不足
>>> 部门经理:权限不足
>>> 总经理:不批

from abc import ABCMeta, abstractmethod
# 模版方法的抽象类:定义了三个插槽方法,等派生类重写
# 定义了一个方法,作为骨架
class Window(metaclass=ABCMeta):
@abstractmethod
def start(self):
pass
@abstractmethod
def repaint(self):
pass
@abstractmethod
def stop(self):
pass
# 骨架方法:模版方法的核心
def run(self):
self.start()
self.repaint()
self.stop()
# 派生类:实现了三个方法,骨架方法在基类中
class MyWindow(Window):
def __init__(self,msg):
self.msg=msg
def start(self):
print("窗口开始运行...")
def stop(self):
print("窗口停止运行...")
def repaint(self):
print(self.msg)
if __name__ == "__main__":
MyWindow("这是窗口信息").run()
#################################
>>> 窗口开始运行...
>>> 这是窗口信息
>>> 窗口停止运行...

from abc import ABCMeta, abstractmethod
# 类定义
class Person:
def __init__(self, name, weight, height):
self.name = name
self.weight = weight
self.height = height
# 定义策略接口:抽象接口类
class Icompare(metaclass=ABCMeta):
@abstractmethod
def compare(self, person1, person2):
pass
# 体重比较策略实现:接口输入和返回的数据类型必须相同
class WeightCompare(Icompare):
def compare(self, person1, person2):
return person1.weight > person2.weight
# 身高比较策略实现:接口输入和返回的数据类型必须相同
class HeightCompare(Icompare):
def compare(self, person1, person2):
return person1.height > person2.height
# 上下文:客户端调用,初始化传入策略对象,
# 对应方法,返回以此策略的结果
class Context:
def __init__(self, strategy):
self.__strategy = strategy
def get_res(self, person1, person2):
return self.__strategy.compare(person1, person2)
if __name__ == "__main__":
xiaoming = Person("xiaoming", 70, 177)
panghu = Person("panghu", 80, 166)
# 客户端选择以哪个算法为比较策略:1、以重量 2、以身高
context = Context(WeightCompare())
# context = Context(HeightCompare())
if context.get_res(xiaoming, panghu):
print("前者比后者大为真")
else:
print("前者比后者大为假")

from abc import ABCMeta, abstractmethod
# 抽象状态类
class State(metaclass=ABCMeta):
@abstractmethod
def rising_temperature(self):
pass
def lowing_temperature(self):
pass
# 具体状态类:以下三个
# 冰块
class SolidState(State):
def rising_temperature(self):
print("温度正在升高...变成液态")
# 链式返回:指定下一个状态的对象
return LiquidState()
def lowing_temperature(self):
print("温度正在降低...已为固态,状态不变")
# 已为终点:返回自己
return self
# 液态水
class LiquidState(State):
def rising_temperature(self):
print("温度正在升高...变成气态")
# 链式返回:指定下一个状态的对象
return GaseousState()
def lowing_temperature(self):
print("温度正在降低...变成固态")
# 链式返回:指定下一个状态的对象
return SolidState()
# 水蒸气
class GaseousState(State):
def rising_temperature(self):
print("温度正在升高...已为气态,状态不变")
# 状态终点:返回自己
return self
def lowing_temperature(self):
print("温度正在降低...变成液态")
# 链式返回:指定下一个状态的对象
return LiquidState()
# 上下文:用来调度状态改变
class Context:
__state = None
# state为类State的派生类对象
def set_state(self, state):
self.__state = state
def rising_temperature(self):
# 类State的派生类对象方法rising_temperature
self.set_state(self.__state.rising_temperature())
def lowing_temperature(self):
# 类State的派生类对象方法lowing_temperature
self.set_state(self.__state.lowing_temperature())
if __name__ == "__main__":
water = Context()
# 先设定状态为气态
water.set_state(GaseousState())
water.rising_temperature()
water.lowing_temperature()
water.lowing_temperature()
water.lowing_temperature()
###############################################
>>> 温度正在升高...已为气态,状态不变
>>> 温度正在降低...变成液态
>>> 温度正在降低...变成固态
>>> 温度正在降低...已为固态,状态不变

# 以下均为具体实现类,之间不互相通信
# 而与中介类一对一通信
class Consumer:
def buy(self):
print("消费者购买产品")
class Producer:
def produce(self):
print("生产者生产产品")
class Mediator:
# 用于保存具体实现类的实例
def __init__(self):
self.consumer = None
self.producer = None
# 此处为核心,各个类之间在此处彼此交互,
# 此处以后也会逐渐膨胀
def deal(self):
self.producer.produce()
self.consumer.buy()
if __name__ == '__main__':
# 创建中介者,并设置两个属性
mediator = Mediator()
mediator.consumer = Consumer()
mediator.producer = Producer()
# 客户端不用关心两个属性的交互过程,被中介者封装了
mediator.deal()

from abc import ABCMeta, abstractmethod
import os
import time
########################################################
# 底层接收者:抽象业务实现类
class ABCReceiver(metaclass=ABCMeta):
@abstractmethod
def create_file(self, filename):
pass
# 回退操作:删除操作固定,所以写死在抽象接收者里
def del_file(self, filename):
os.remove(filename)
# 具体业务实现类:接收命令对象传来的参数,其为扩展点,可以扩展具体如何创建
class FileReceiver(ABCReceiver):
def create_file(self, filename):
with open(filename, 'w')as f:
pass
########################################################
# 中层命令接口:核心,调用与实现的中间层,解耦,接收传入的接收者对象和参数
class AbcCommand(metaclass=ABCMeta):
@abstractmethod
def execute(self):
pass
# 回退操作
@abstractmethod
def cancel(self):
pass
# 中层具体命令:调用具体接收者实现功能
class FileCommand(AbcCommand):
def __init__(self, receiver, new_file_name):
self.receiver = receiver
self.new_file_name = new_file_name
def execute(self):
self.receiver.create_file(self.new_file_name)
# 回退操作:多命令是先回复最后执行命令并依次向前
def cancel(self):
self.receiver.del_file(self.new_file_name)
########################################################
# 顶层调用者:所有操作都被包含在调用者对象中,生成一次对象,后续运行
# 这一组封装命令只需调用run方法即可
# 原子化操作:操作要么所有命令都发生,要么全都不发生(出现问题回滚命令)
class FileInvoker:
def __init__(self, in_command):
self.command = in_command
def run(self):
self.command.execute()
# 回退操作
def cannel(self):
self.command.cancel()
########################################################
if __name__ == '__main__':
# 接收者对象:此处可以选择不同接收者,以选择不同命令组合
new_file_receiver = FileReceiver()
# 命令对象:传入接收者对象和参数
command = FileCommand(new_file_receiver, 'data.txt')
# 构造调用者:传入命令对象
invoker = FileInvoker(command)
# 执行命令
invoker.run()
# 暂停三秒
time.sleep(3)
# 现在又想撤销
invoker.cannel()
# 暂停三秒
time.sleep(3)
# 又想重复执行
invoker.run()

# 发起者:至少需要实现save和restore两个方法
class Machine:
def __init__(self, context):
self._state = context
def get_context(self):
print("当前状态为:", self._state)
def set_context(self, context):
self._state = context
# 此处需要手动创建需要保存的字典,传给备忘录对象
def save(self):
print("保存中...保存完毕")
return Memento({'state': self._state})
# 从备忘录中拿到状态并提取其中的值覆盖发起者的属性
def restore(self, memento):
print("读取中...读取完毕")
self._state = memento.state['state']
# 备忘录类:只包含状态属性
class Memento:
def __init__(self, state):
self.state = state
# 备忘录管理员类:负责储存历次save的备忘录
class Caretaker:
def __init__(self, machine):
# 备忘录保存池
self._mementos = []
# 为了self.save方法调用发起者save方法
self._originator = machine
# 将发起者生成的备忘录类实例保存在本备忘录保存池中
def save(self):
self._mementos.append(self._originator.save())
# 从备忘录池后面取备忘录给发起者进行恢复
def restore(self):
# 加了判断:如果空,直接返回None
if not len(self._mementos):
print("备忘录中已无备份")
return
memento = self._mementos.pop()
self._originator.restore(memento)
if __name__ == "__main__":
# 客户端:生成发起类对象、备忘录管理对象
originator = Machine("初始化状态")
caretaker = Caretaker(originator)
# 隐式生成备忘录对象及添加到备忘录管理对象的备忘录池中
caretaker.save()
originator.set_context("第一次修改")
caretaker.save()
caretaker.restore()
originator.get_context()
caretaker.restore()
originator.get_context()
print(20*"#")
caretaker.restore()
originator.get_context()
############################################
>>> 保存中...保存完毕
>>> 保存中...保存完毕
>>> 读取中...读取完毕
>>> 当前状态为: 第一次修改
>>> 读取中...读取完毕
>>> 当前状态为: 初始化状态
>>> ####################
>>> 备忘录中已无备份
>>> 当前状态为: 初始化状态

from abc import abstractmethod, ABCMeta
######################################################
# 数据节点:数据对象
class Product(metaclass=ABCMeta):
# 此处为数据:特点是已经稳定,不会增加或减少
def __init__(self, name, price):
self.name = name
self.price = price
# 核心方法:传入访问者对象
@abstractmethod
def accept(self, visitor):
pass
# accept方法会调用访问者对象的visit方法
class Food(Product):
def accept(self, visitor):
visitor.visit(self)
class Drink(Product):
def accept(self, visitor):
visitor.visit(self)
######################################################
# 访问者对象:依赖于数据节点对象
class Visitor(metaclass=ABCMeta):
# 核心方法:传入数据节点对象
@abstractmethod
def visit(self, product):
pass
class Seller:
# 此处接收数据节点对象,并提取其中的数据,通常是读操作
def visit(self, product):
print("出售者查看了产品的价格:", product.price)
class Buyer:
# 两个访问者会进行不同的数据处理,不会影响原数据节点对象
def visit(self, product):
print("购买者查看了产品的名称:", product.name)
######################################################
# 对象结构:用于存储数据节点对象的数据池
class ProductPile:
# 存储生成的数据节点对象们
def __init__(self):
self.products = []
# 向数据池中增加对象
def add_product(self, obj):
self.products.append(obj)
def remove_products(self, obj):
self.products.remove(obj)
# 数据池遍历每个数据节点并调用 其accept方法
# 进而调用其中 访问者的visit方法
def visit(self, visitor):
for obj in self.products:
obj.accept(visitor)
if __name__ == '__main__':
# 创建节点池,并新增节点
product_pool = ProductPile()
product_pool.add_product(Food("面包", 5))
product_pool.add_product(Food("火腿肠", 2))
product_pool.add_product(Drink("矿泉水", 2))
# 创建访问者角色并传给数据池
seller = Seller()
buyer = Buyer()
product_pool.visit(seller)
product_pool.visit(buyer)
######################################################
>>> 出售者查看了产品的价格: 5
>>> 出售者查看了产品的价格: 2
>>> 出售者查看了产品的价格: 2
>>> 购买者查看了产品的名称: 面包
>>> 购买者查看了产品的名称: 火腿肠
>>> 购买者查看了产品的名称: 矿泉水