Python教程 Episode 22 - Python 设计模式实战

欢迎来到"Python教程从零基础到实战"系列的第二十二期！

经过了前面二十一期，你已经掌握了 Python 的基础语法、数据结构、面向对象编程、爬虫、数据分析、Web 开发、部署运维和性能优化。但你有没有遇到过这样的代码——

```python

# 一个"功能完善"但结构混乱的配置管理模块

defget_config(key):

"""获取配置"""

# 先读环境变量

    val = os.environ.get(key)

if val:

return val

# 再读配置文件

# ...一堆 if/else...

# 再读数据库...

# ...又一堆逻辑...

return default

defset_config(key, value):

# 又要写环境变量、写文件、写数据库...

# 复制粘贴了无数行 get_config 的逻辑

pass

```

这段代码能跑，但维护起来非常痛苦。改一个逻辑要动到处函数，加一个新配置来源要改每一处。问题出在哪？**没有好的架构设计**。

这一期，我们来讲——**设计模式**。不照搬 textbook，我们用真实的 Python 项目来说明，每种模式怎么解决实际问题。

---

## 一、为什么需要设计模式？

### 1.1 设计模式的本质

设计模式不是"必须遵守的规则"，而是**前辈们总结的、经过验证的解决方案模板**。它们回答的问题是：

> "当 A 情况发生时，用 B 方式来解决，能避免 C 问题。"

比如：**多个类需要共享同一个全局状态**——用**单例模式**，比在每次调用时手动判断"我是不是第一个创建的实例"要优雅得多。

### 1.2 Python 的特殊性

和其他语言不同，Python 有几个独特的优势：

-**一切皆对象**：函数也是对象，可以传递、返回、存储

-**动态特性**：可以在运行时修改类的行为

-**装饰器**：天然支持"在不修改源码的前提下增强功能"

这意味着很多在其他语言里需要用复杂套路实现的模式，在 Python 里有更简单的写法。但这不意味着设计模式没用——它们提供的**结构化思路**依然不可替代。

---

## 二、创建型模式：如何优雅地创建对象

### 2.1 单例模式（Singleton）

**场景**：配置管理器、日志器、数据库连接池——整个程序只需要一个实例。

#### 方案一：模块本身就是单例

```python

# config.py — Python 模块导入天然就是单例！

import os

classConfigManager:

"""配置管理器：确保全局只有一个实例"""

    _instance = None

def__new__(cls):

ifcls._instance isNone:

print("[Config] 创建新实例")

cls._instance = super().__new__(cls)

cls._instance._initialized = False

returncls._instance

def__init__(self):

ifself._initialized:

return# 已经初始化过了，跳过

self._initialized = True

self._config = {}

self._load_from_env()

self._load_from_file()

def_load_from_env(self):

"""从环境变量加载"""

        env_vars = {

"DATABASE_URL": os.getenv("DATABASE_URL"),

"API_KEY": os.getenv("API_KEY"),

"DEBUG": os.getenv("DEBUG", "false").lower() == "true",

        }

for key, value in env_vars.items():

if value isnotNone:

self._config[key] = value

def_load_from_file(self):

"""从 YAML 配置文件加载"""

try:

import yaml

withopen("config.yaml", "r", encoding="utf-8") as f:

                file_config = yaml.safe_load(f)

if file_config:

self._config.update(file_config)

exceptFileNotFoundError:

pass# 没有配置文件就跳过

exceptImportError:

# yaml 库不存在时使用 JSON

import json

try:

withopen("config.json", "r", encoding="utf-8") as f:

self._config.update(json.load(f))

exceptFileNotFoundError:

pass

defget(self, key, default=None):

returnself._config.get(key, default)

defset(self, key, value):

self._config[key] = value

def__repr__(self):

returnf"ConfigManager(config={dict(self._config)})"

# 使用

cfg1 = ConfigManager()

cfg2 = ConfigManager()

print(cfg1 is cfg2)  # True! 它们是同一个实例

cfg1.set("APP_NAME", "MyApp")

print(cfg2.get("APP_NAME"))  # "MyApp" — 通过另一个实例也能访问

```

#### 方案二：用装饰器实现

```python

from functools import wraps

defsingleton(cls):

"""装饰器版本：适用于任意类"""

    instances = {}

@wraps(cls)

defget_instance(*args, **kwargs):

ifclsnotin instances:

            instances[cls] = cls(*args, **kwargs)

return instances[cls]

return get_instance

@singleton

classDatabaseConnection:

def__init__(self, url: str):

self.url = url

print(f"[DB] 连接到 {url}")

defquery(self, sql: str) -> list:

print(f"[DB] 执行 SQL: {sql}")

return [{"id": 1, "name": "Alice"}]

db1 = DatabaseConnection("postgresql://localhost/mydb")

db2 = DatabaseConnection("postgresql://localhost/otherdb")  # ← 被忽略！

print(db1.url)  # "postgresql://localhost/mydb" — 参数只接受第一次的

```

#### 方案三：Pythonic 的做法（推荐）

```python

# 实际上，对于纯配置场景，最 Pythonic 的方式根本不需要单例：

# 直接用模块级别的变量就行

# settings.py

DATABASE_URL = "postgresql://localhost/mydb"

API_KEY = "sk-xxxxx"

DEBUG = False

# 在其他文件中直接导入

from settings importDATABASE_URL, DEBUG

# 或者作为命名空间

import settings

print(settings.DATABASE_URL)

```

**经验法则**：Python 中，如果功能能用模块级别变量实现，就别用单例。单例的真正战场在——需要"状态"的场景，比如日志器、指标采集器。

---

### 2.2 工厂模式（Factory）

**场景**：需要根据用户输入或配置，动态选择创建哪种类型的对象。

```python

from abc importABC, abstractmethod

import json

import yaml

classFormatter(ABC):

"""输出格式化器的抽象基类"""

@abstractmethod

defformat(self, data: dict) -> str:

pass

classJSONFormatter(Formatter):

defformat(self, data: dict) -> str:

return json.dumps(data, ensure_ascii=False, indent=2)

classCSVFormatter(Formatter):

defformat(self, data: dict) -> str:

ifnot data:

return""

        headers = list(data[0].keys())

        lines = [",".join(headers)]

for row in data:

            lines.append(",".join(str(row[h]) for h in headers))

return"\n".join(lines)

classMarkdownTableFormatter(Formatter):

defformat(self, data: dict) -> str:

ifnot data:

return""

        headers = list(data[0].keys())

        lines = ["| " + " | ".join(headers) + " |"]

        lines.append("| " + " | ".join("---"for _ in headers) + " |")

for row in data:

            lines.append("| " + " | ".join(str(row[h]) for h in headers) + " |")

return"\n".join(lines)

# ===== 工厂：负责创建正确的格式化器 =====

classFormatterFactory:

"""集中管理格式化器的创建"""

# 注册表：格式名 -> 格式化器类

    _formatters: dict[str, type] = {

"json": JSONFormatter,

"csv": CSVFormatter,

"markdown": MarkdownTableFormatter,

    }

@classmethod

defregister(cls, name: str, formatter_class: type):

"""允许外部注册新的格式化器"""

cls._formatters[name] = formatter_class

@classmethod

defcreate(cls, fmt: str) -> Formatter:

"""根据名字创建格式化器"""

if fmt notincls._formatters:

raiseValueError(

f"未知的格式 '{fmt}'，支持的格式: {list(cls._formatters.keys())}"

            )

returncls._formatters[fmt]()

# 使用示例

users = [

    {"id": 1, "name": "张三", "role": "管理员"},

    {"id": 2, "name": "李四", "role": "普通用户"},

]

# 创建不同格式的输出

for fmt_type in ["json", "csv", "markdown"]:

    formatter = FormatterFactory.create(fmt_type)

print(f"\n=== {fmt_type.upper()} ===")

print(formatter.format(users))

# 扩展：用户可以注册自己的格式化器

classXMLFormatter(Formatter):

defformat(self, data: dict) -> str:

        lines = ["<?xml version='1.0' encoding='utf-8'?>", "<records>"]

for record in data:

            line = "  <record>"

for key, value in record.items():

                line += f"<{key}>{value}</{key}>"

            line += "</record>"

            lines.append(line)

        lines.append("</records>")

return"\n".join(lines)

FormatterFactory.register("xml", XMLFormatter)

xml_formatter = FormatterFactory.create("xml")

print(f"\n=== xml ===\n{xml_formatter.format(users)}")

```

---

### 2.3 建造者模式（Builder）

**场景**：构建一个复杂的 HTTP 请求、构建一份包含多个选项的报告、构建一个数据库查询。

```python

classHttpRequestBuilder:

"""链式调用的 HTTP 请求构建器"""

def__init__(self, method: str, url: str):

self.method = method

self.url = url

self._headers: dict[str, str] = {}

self._params: dict[str, str] = {}

self._body = None

defheader(self, key: str, value: str) -> "HttpRequestBuilder":

"""添加请求头——返回 self 以支持链式调用"""

self._headers[key] = value

returnself# ← 关键：返回自身

defparams(self, **kwargs) -> "HttpRequestBuilder":

"""添加 URL 查询参数"""

self._params.update(kwargs)

returnself

defjson_body(self, data: dict) -> "HttpRequestBuilder":

"""设置 JSON body"""

import json

self._body = json.dumps(data).encode("utf-8")

self.header("Content-Type", "application/json")

returnself

deftext_body(self, text: str) -> "HttpRequestBuilder":

"""设置纯文本 body"""

self._body = text.encode("utf-8")

self.header("Content-Type", "text/plain")

returnself

defbuild(self) -> dict:

"""构建最终请求"""

from urllib.parse import urlencode

        request = {

"method": self.method,

"url": self.url,

"headers": self._headers,

"body": self._body,

        }

ifself._params:

            query_string = urlencode(self._params)

            request["url"] += f"?{query_string}"

return request

# ===== 使用示例 =====

# 链式调用构建 GET 请求

get_request = HttpRequestBuilder("GET", "https://api.example.com/users").params(

page=1, per_page=10

).build()

print("GET 请求:", get_request)

# 链式调用构建 POST 请求

post_request = HttpRequestBuilder("POST", "https://api.example.com/users").header(

"Authorization", "Bearer my-token"

).json_body({"name": "王五", "email": "wang@example.com"}).build()

print("\nPOST 请求:", post_request)

# 复杂请求

complex_request = (

    HttpRequestBuilder("PUT", "https://api.example.com/users/42")

    .header("Authorization", "Bearer my-token")

    .header("If-Match", '"v1"')

    .json_body({"name": "王五 updated"})

    .build()

)

print("\nPUT 请求:", complex_request)

```

**什么时候用建造者？** 当你需要构建的对象有很多可选参数时，建造者比一堆可选参数的构造函数要清晰得多。

---

## 三、结构型模式：如何优雅地组合对象

### 3.1 适配器模式（Adapter）

**场景**：你的代码想用一个 API，但供应商提供了不同的接口。适配器帮你"翻译"。

```python

from abc importABC, abstractmethod

import smtplib

from email.mime.text import MIMEText

classEmailService(ABC):

"""我们的代码所期望的统一邮件服务接口"""

@abstractmethod

defsend(self, to: str, subject: str, body: str) -> bool:

pass

classSMTPAdapter(EmailService):

"""适配器：将统一接口适配到 SMTP 协议"""

def__init__(self, host: str, port: int, username: str, password: str):

self.host = host

self.port = port

self.username = username

self.password = password

defsend(self, to: str, subject: str, body: str) -> bool:

        msg = MIMEText(body, "html", "utf-8")

        msg["Subject"] = subject

        msg["From"] = self.username

        msg["To"] = to

try:

            server = smtplib.SMTP(self.host, self.port)

            server.starttls()

            server.login(self.username, self.password)

            server.sendmail(self.username, [to], msg.as_string())

            server.quit()

returnTrue

exceptExceptionas e:

print(f"SMTP 发送失败: {e}")

returnFalse

classSendGridAdapter(EmailService):

"""适配器：将统一接口适配到 SendGrid API"""

def__init__(self, api_key: str):

self.api_key = api_key

defsend(self, to: str, subject: str, body: str) -> bool:

import requests

        data = {

"personalizations": [{"to": [{"email": to}]}],

"from": {"email": "noreply@example.com"},

"subject": subject,

"content": [{"type": "text/html", "value": body}],

        }

try:

            resp = requests.post(

"https://api.sendgrid.com/v3/mail/send",

headers={

"Authorization": f"Bearer {self.api_key}",

"Content-Type": "application/json",

                },

json=data,

timeout=10,

            )

return resp.status_code == 202

exceptExceptionas e:

print(f"SendGrid 发送失败: {e}")

returnFalse

classAWSSESAdapter(EmailService):

"""适配器：适配 AWS SES"""

def__init__(self, region: str, access_key: str, secret_key: str):

self.region = region

self.access_key = access_key

self.secret_key = secret_key

defsend(self, to: str, subject: str, body: str) -> bool:

try:

import boto3

            ses = boto3.client("ses", region_name=self.region)

            ses.send_email(

Source=self.access_key + "@example.com",

Destination={"ToAddresses": [to]},

Message={

"Subject": {"Data": subject},

"Body": {"Html": {"Data": body}},

                },

            )

returnTrue

exceptExceptionas e:

print(f"AWS SES 发送失败: {e}")

returnFalse

# ===== 使用示例：上层业务代码完全不用关心底层实现 =====

defnotify_user(email_service: EmailService, user_email: str, message: str):

"""通知用户——只依赖抽象接口"""

    success = email_service.send(user_email, "系统通知", message)

if success:

print(f"  ✓ 通知已发送至 {user_email}")

else:

print(f"  ✗ 通知发送失败: {user_email}")

# 可以随时切换邮件服务，不需要修改 notify_user 的代码！

smtp_service = SMTPAdapter("smtp.gmail.com", 587, "user@gmail.com", "password")

notify_user(smtp_service, "admin@example.com", "<h1>服务器健康</h1><p>一切正常</p>")

# 切换到 SendGrid

sendgrid_service = SendGridAdapter("SG.xxxxx")

notify_user(sendgrid_service, "admin@example.com", "<h1>服务器健康</h1><p>一切正常</p>")

```

---

### 3.2 装饰器模式（Decorator）

你已经用过 Python 的 `@` 装饰器语法了，但那是**函数装饰器**——它修改的是函数本身。装饰器模式修饰的是**对象的行为**，两者概念不同但思路相通。

```python

from abc importABC, abstractmethod

classCoffee(ABC):

"""咖啡抽象类"""

@abstractmethod

defcost(self) -> float:

pass

@abstractmethod

defdescription(self) -> str:

pass

classSimpleCoffee(Coffee):

"""基础咖啡"""

defcost(self) -> float:

return10.0

defdescription(self) -> str:

return"简易咖啡"

# ===== 装饰器：给咖啡加料 =====

classCoffeeDecorator(Coffee):

"""咖啡装饰器基类"""

def__init__(self, coffee: Coffee):

self._coffee = coffee

defcost(self) -> float:

returnself._coffee.cost()

defdescription(self) -> str:

returnself._coffee.description()

classMilk(CoffeeDecorator):

"""牛奶装饰器"""

defcost(self) -> float:

returnself._coffee.cost() + 3.0

defdescription(self) -> str:

returnf"{self._coffee.description()} + 牛奶"

classWhippedCream(CoffeeDecorator):

"""奶油装饰器"""

defcost(self) -> float:

returnself._coffee.cost() + 2.0

defdescription(self) -> str:

returnf"{self._coffee.description()} + 奶油"

classSugar(CoffeeDecorator):

"""糖装饰器"""

defcost(self) -> float:

returnself._coffee.cost() + 1.0

defdescription(self) -> str:

returnf"{self._coffee.description()} + 糖"

# ===== 使用示例 =====

my_coffee = SimpleCoffee()

print(f"{my_coffee.description()}: ¥{my_coffee.cost():.1f}")

# 简易咖啡: ¥10.0

my_coffee = Milk(SimpleCoffee())

print(f"{my_coffee.description()}: ¥{my_coffee.cost():.1f}")

# 简易咖啡 + 牛奶: ¥13.0

my_coffee = WhippedCream(Milk(SimpleCoffee()))

print(f"{my_coffee.description()}: ¥{my_coffee.cost():.1f}")

# 简易咖啡 + 牛奶 + 奶油: ¥15.0

my_coffee = Sugar(WhippedCream(Milk(SimpleCoffee())))

print(f"{my_coffee.description()}: ¥{my_coffee.cost():.1f}")

# 简易咖啡 + 牛奶 + 奶油 + 糖: ¥16.0

```

Python 的函数装饰器和对象装饰器模式可以结合使用：

```python

from functools import wraps

import time

deftimer(func):

"""计时装饰器——测量函数执行时间"""

@wraps(func)

defwrapper(*args, **kwargs):

        start = time.perf_counter()

        result = func(*args, **kwargs)

        elapsed = time.perf_counter() - start

print(f"  [{func.__name__}] 耗时 {elapsed:.4f}s")

return result

return wrapper

defretry(max_retries: int = 3, delay: float = 1.0):

"""重试装饰器——失败时自动重试"""

defdecorator(func):

@wraps(func)

defwrapper(*args, **kwargs):

for attempt inrange(1, max_retries + 1):

try:

return func(*args, **kwargs)

exceptExceptionas e:

if attempt == max_retries:

raise

print(f"  ⚠ 第 {attempt} 次尝试失败: {e}，{delay}s 后重试")

                    time.sleep(delay)

return wrapper

return decorator

# 组合使用

@timer

@retry(max_retries=3, delay=0.1)

defunstable_api_call(request_id: int) -> dict:

"""模拟一个偶尔会失败的 API 调用"""

import random

if random.random() < 0.3:

raiseConnectionError("网络连接不稳定")

return {"request_id": request_id, "data": "success"}

result = unstable_api_call(42)

print(f"  结果: {result}")

```

---

### 3.3 外观模式（Facade）

**场景**：子系统太复杂，提供一个简化的统一入口。

```python

import subprocess

import os

from datetime import datetime

classDockerOperations:

"""Docker 操作门面——把复杂的 docker 命令封装成简单方法"""

def__init__(self, project_dir: str = "."):

self.project_dir = project_dir

defbuild_image(self, image_name: str, tag: str = "latest") -> bool:

"""构建镜像"""

        cmd = f'docker build -t "{image_name}:{tag}" .'

print(f"  🔨 构建镜像: {cmd}")

        result = subprocess.run(cmd, shell=True, capture_output=True, text=True,

cwd=self.project_dir)

return result.returncode == 0

defrun_container(self, image_name: str, container_name: str,

ports: dict = None, detach: bool = True) -> bool:

"""运行容器"""

        cmd = f'docker run'

if detach:

            cmd += " -d"

if container_name:

            cmd += f' --name "{container_name}"'

if ports:

for host, container in ports.items():

                cmd += f' -p {host}:{container}'

        cmd += f' "{image_name}:latest"'

print(f"  ▶ 启动容器: {cmd}")

        result = subprocess.run(cmd, shell=True, capture_output=True, text=True,

cwd=self.project_dir)

return result.returncode == 0

defstop_container(self, container_name: str) -> bool:

"""停止容器"""

        cmd = f'docker stop "{container_name}"'

print(f"  ⏸ 停止容器: {cmd}")

        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)

return result.returncode == 0

defremove_container(self, container_name: str) -> bool:

"""删除容器"""

        cmd = f'docker rm "{container_name}"'

print(f"  🗑 删除容器: {cmd}")

        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)

return result.returncode == 0

defhealth_check(self, container_name: str) -> dict:

"""检查容器状态"""

import json

        cmd = f'docker inspect --format="{{{{json .State}}}}" "{container_name}"'

        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)

if result.returncode != 0:

return {"status": "not_found"}

        state = json.loads(result.stdout)

return {

"running": state.get("Running", False),

"status": state.get("Status", "unknown"),

"exit_code": state.get("ExitCode", -1),

"started_at": state.get("StartedAt", ""),

        }

# ===== 外观：一行搞定整个部署流程 =====

classDeploymentFacade:

"""部署外观——对外只暴露一个 deploy() 方法"""

def__init__(self, app_name: str, project_dir: str = "."):

self.app_name = app_name

self.docker = DockerOperations(project_dir)

self.tag = datetime.now().strftime("%Y%m%d-%H%M%S")