数据STUDIO

这 7 个 Python 设计模式让你效率翻倍

Image

“上周帮团队review一段订单处理代码,密密麻麻的if-else判断支付渠道、导出格式、通知方式……新加一个微信支付要改5处地方,单元测试直接挂了4个。”

你是不是也遇到过类似的场景?代码能跑,但每次加功能都像拆弹。

其实,设计模式不是面试八股文——7个Python原生支持的套路,就能把“意大利面式代码”变成清晰可扩展的工程。读完这篇,你也能写出同事直呼“优雅”的Python代码。

1. 单例(Singleton):全局唯一的“独生子”

类比:公司只有一台打印机,所有人共享使用。谁去打印都拿到同一台机器,不会出现“A加纸了,B却拿到空机器”的混乱。

正式定义:确保一个类只有一个实例,并提供全局访问点。

代码实战(Python 3.8+)

classDatabase:
    _instance = None

def__new__(cls):
if cls._instance isNone:
            cls._instance = super(Database, cls).__new__(cls)
            cls._instance.connection = "Connected to DB"# 模拟连接
return cls._instance

# 验证
db1 = Database()
db2 = Database()
print(db1 is db2)  # True,同一个对象
print(db1.connection)  # Connected to DB

⚠️ 注意:90%的人会踩坑
单例在多线程环境下可能失效(多个线程同时进入if cls._instance is None)。生产环境请使用threading.Lock或元类实现。另外,单元测试会因状态泄漏而互相影响——每个测试用例需要重置实例,否则前面的mock会污染后面的断言。

适用场景:数据库连接池、全局配置对象、日志记录器。
不适用:异步任务、多线程频繁读写的状态。

2. 工厂方法(Factory Method):生产对象的“流水线”

类比:你去快餐店点餐,告诉店员“要汉堡”,他不会问你“要哪种面包、肉饼怎么做”,而是直接给你对应的汉堡。你不需要知道生产细节。

正式定义:定义一个创建对象的接口,但由子类决定实例化哪一个类。

代码实战

classExporter:
"""导出器基类"""
defexport(self, data):
raise NotImplementedError

classPDFExporter(Exporter):
defexport(self, data):
returnf"将 {data} 导出为 PDF"

classCSVExporter(Exporter):
defexport(self, data):
returnf"将 {data} 导出为 CSV"

# 工厂函数
defexporter_factory(kind: str) -> Exporter:
if kind == "pdf":
return PDFExporter()
elif kind == "csv":
return CSVExporter()
else:
raise ValueError(f"不支持格式: {kind}")

# 使用
exporter = exporter_factory("pdf")
print(exporter.export("年度报表"))  # 将 年度报表 导出为 PDF

复杂度:时间O(1),空间O(1)(不计对象内部数据)。

扩展思考:国内报表系统常用POI导出Excel,可以用工厂模式统一ExcelExporter。新加一种格式(如JSON)只需新增类,不必修改已有的if-else(开闭原则)。但Python中如果类型很少,直接用字典映射更简单:

exporters = {"pdf": PDFExporter, "csv": CSVExporter}
exporter = exporters.get("pdf", PDFExporter)()

3. 观察者(Observer):事件触发的“广播系统”

类比:你关注了B站UP主,每次他发视频,你和其他粉丝都会收到通知。你不需要轮询“他发新视频了吗”,而是UP主主动推给你。

正式定义:定义一对多的依赖关系,当一个对象状态改变时,所有依赖它的对象都会自动收到通知。

代码实战

classOrderEvent:
"""订单事件中心"""
def__init__(self):
        self.subscribers = []  # 订阅者列表

defsubscribe(self, callback):
        self.subscribers.append(callback)

defnotify(self, order_data):
for callback in self.subscribers:
            callback(order_data)

# 定义两个观察者行为
defsend_email(order):
    print(f"发送邮件通知:订单 {order} 已创建")

defupdate_analytics(order):
    print(f"更新数据看板:订单 {order} 计入统计")

# 使用
event = OrderEvent()
event.subscribe(send_email)
event.subscribe(update_analytics)

event.notify("Order#123")
# 输出:
# 发送邮件通知:订单 Order#123 已创建
# 更新数据看板:订单 Order#123 计入统计

⚠️ 调试噩梦:当你有十几个订阅者时,一个notify调用会触发一连串行为,堆栈很难追踪。建议给每个回调加唯一ID,并在日志中记录调用链。

适用场景:消息队列、Webhook、GUI事件响应。国内常用RocketMQ或Redis Pub/Sub实现分布式观察者。

4. 策略(Strategy):可插拔的“算法切换器”

类比:打车回家,可以选“快车”“专车”“拼车”。计价算法不同,但你的操作(输入起点终点)完全一样。策略就是“选哪个算法”。

正式定义:定义一系列算法,将每个算法封装起来,并使它们可以互相替换。

代码实战(对比错误与正确写法)

❌ 错误做法:硬编码if-else

defget_price(base_price, strategy_type):
if strategy_type == "discount":
return base_price * 0.9
elif strategy_type == "surge":
return base_price * 1.5
# 每次加策略都要改这个函数

✅ 最佳实践:策略类 + 运行时替换

from abc import ABC, abstractmethod

classPricingStrategy(ABC):
    @abstractmethod
defcalculate(self, price):
pass

classDiscountStrategy(PricingStrategy):
defcalculate(self, price):
return price * 0.9

classSurgeStrategy(PricingStrategy):
defcalculate(self, price):
return price * 1.5

classPriceCalculator:
def__init__(self, strategy: PricingStrategy):
        self.strategy = strategy

defget_price(self, base_price):
return self.strategy.calculate(base_price)

# 使用
calc = PriceCalculator(DiscountStrategy())
print(calc.get_price(100))  # 90.0

# 运行时切换策略
calc.strategy = SurgeStrategy()
print(calc.get_price(100))  # 150.0

复杂度:每个策略O(1),替换策略O(1)。

国内实践:双十一的“满减策略”“限时折扣策略”“会员价策略”都适合用策略模式。但注意Python中可以直接传函数对象(一等公民),不一定非要写类:

defdiscount(price):return price * 0.9
defsurge(price):return price * 1.5

classPriceCalculator:
def__init__(self, strategy_func):
        self.strategy = strategy_func

5. 装饰器(Decorator):无侵入的“能力增强器”

类比:手机贴膜——不改变手机内部电路,但增加了防摔功能。装饰器就是在不修改原函数代码的前提下,附加新能力。

正式定义:动态地给一个对象添加一些额外的职责。

代码实战(FastAPI鉴权场景)

defauth_required(fn):
"""鉴权装饰器:只有admin可以调用"""
defwrapper(user, *args, **kwargs):
if user != "admin":
raise PermissionError(f"用户 {user} 无权限")
return fn(user, *args, **kwargs)
return wrapper

@auth_required
defdelete_user(user, user_id):
    print(f"{user} 删除了用户 {user_id}")

# 正确调用
delete_user("admin", 42)   # admin 删除了用户 42

# 错误调用(会抛出异常)
# delete_user("guest", 42)  # PermissionError

复杂度:装饰器本身O(1),但多层嵌套(如@auth @log @cache)会增加调用栈深度,调试时需用functools.wraps保留原函数元数据。

⚠️ 注意:装饰器执行顺序是从下往上(靠近函数的先执行)。另外,装饰器内的wrapper如果不返回原函数返回值,会吞掉结果。正确的装饰器应该写成:

from functools import wraps
defauth_required(fn):
    @wraps(fn)
defwrapper(user, *args, **kwargs):
        ...
return fn(user, *args, **kwargs)
return wrapper

Python原生支持:@staticmethod、@classmethod、@property都是装饰器。国内常用于FastAPI依赖注入、Django权限校验、日志埋点。

6. 命令(Command):延迟执行的“任务卡”

类比:餐厅里服务员写下订单小票(命令对象),后厨按顺序执行,不需要知道这道菜是谁点的。支持撤销、重试、排队。

正式定义:将请求封装为对象,从而支持参数化、队列、日志和撤销操作。

代码实战(后台任务队列)

from abc import ABC, abstractmethod

classCommand(ABC):
    @abstractmethod
defexecute(self):
pass

classSendInvoiceCommand(Command):
defexecute(self):
        print("发送发票邮件")

classResizeImageCommand(Command):
defexecute(self):
        print("压缩图片尺寸")

# 构建任务队列
queue = []
queue.append(SendInvoiceCommand())
queue.append(ResizeImageCommand())

# 统一执行
for job in queue:
    job.execute()

扩展:带重试的命令

classRetryCommand(Command):
def__init__(self, cmd, max_retries=3):
        self.cmd = cmd
        self.retries = max_retries

defexecute(self):
for i in range(self.retries):
try:
                self.cmd.execute()
break
except Exception as e:
                print(f"重试 {i+1} 失败: {e}")

适用场景:Celery任务、撤销/重做(如Ctrl+Z)、宏命令录制。国内常用APScheduler + 命令模式实现延迟任务。

7. 适配器(Adapter):接口不兼容的“转换插头”

类比:你带了一个三脚插头(PayPal接口),但墙上只有两孔插座(Razorpay接口)。适配器(转换插头)让两者能合作。

正式定义:将一个类的接口转换成客户希望的另一个接口,使原本不兼容的类能一起工作。

代码实战(统一支付网关)

# 两个第三方SDK,接口不同
classPayPal:
defsend_payment(self, amount):
        print(f"PayPal 支付 {amount} 元")

classAlipay:
deftransfer(self, amount):
        print(f"支付宝支付 {amount} 元")

# 适配器:统一为 pay(amount)
classPaymentAdapter:
def__init__(self, provider):
        self.provider = provider

defpay(self, amount):
if hasattr(self.provider, 'send_payment'):
            self.provider.send_payment(amount)
elif hasattr(self.provider, 'transfer'):
            self.provider.transfer(amount)
else:
raise TypeError("不支持的支付提供方")

# 业务代码只依赖 pay 方法
defcheckout(adapter: PaymentAdapter, amount):
    adapter.pay(amount)

# 使用
paypal_adapter = PaymentAdapter(PayPal())
alipay_adapter = PaymentAdapter(Alipay())

checkout(paypal_adapter, 100)  # PayPal 支付 100 元
checkout(alipay_adapter, 200)  # 支付宝支付 200 元

⚠️ 注意:适配器不是万能胶。如果两个接口语义相差太大(比如一个需要异步回调,另一个是同步返回),强行适配会引入诡异bug。此时应考虑封装整个第三方调用为统一服务层,而不是逐个方法适配。

国内常见场景:对接微信支付、支付宝、银联云闪付;或统一不同云厂商的OSS接口(阿里云OSS vs 腾讯云COS)。

写在最后

设计模式不是为了用而用。我见过有人为了“优雅”在10行代码里塞进3个模式,结果新人接手直接自闭。过度设计的危害不亚于面条代码。一个简单的经验:当你第三次写相似的if-else或复制粘贴相同逻辑时,再考虑引入模式。

另外,Python的鸭子类型和一等函数已经简化了很多经典模式(比如策略可以直接传lambda,观察者可以用Event库)。不要强行套用Java风格的类图,写出“Pythonic”的模式才是真本事。

记忆点回顾

  1. 单例:全局唯一实例,注意线程安全和测试隔离。
  2. 工厂方法:将对象创建与使用分离,扩展新类型无需修改旧代码。
  3. 观察者:事件驱动解耦,但调试困难,适合一对多通知。
  4. 策略:运行时切换算法,Python中可直接传函数简化。
  5. 装饰器:Python原生支持,无侵入增强函数,注意顺序和wraps。
  6. 命令:封装操作为对象,支持队列、重试、撤销。
  7. 适配器:统一异构接口,接入第三方库时必备。

你在项目中用过哪个设计模式解决了实际问题?或者踩过什么“过度设计”的坑?欢迎在评论区分享

🏴‍☠️宝藏级🏴‍☠️ 原创公众号『数据STUDIO』内容超级硬核。公众号以Python为核心语言,垂直于数据科学领域,包括可戳👉Python|MySQL|数据分析|数据可视化|机器学习与数据挖掘|爬虫等,从入门到进阶!

长按👇关注- 数据STUDIO -设为星标,干货速递ImageImage