首页
/ 彻底重构！从面条代码到工业级Python架构：betterpython项目实战指南

彻底重构！从面条代码到工业级Python架构：betterpython项目实战指南

2026-01-15 17:01:01作者：魏献源Searcher
betterpython
Code examples for my Write Better Python Code series on YouTube.
项目地址：https://gitcode.com/gh_mirrors/be/betterpython

你是否还在为混乱的Python代码库头疼？函数超过500行、类之间耦合紧密、修改一处牵一发而动全身？本文将通过分析GitHub星标项目betterpython的10大重构案例，带你掌握从代码整洁到架构设计的完整方法论。读完本文，你将获得：

  • 识别代码坏味道的7个实用技巧
  • 5大设计模式的Pythonic实现方案
  • SOLID原则的逐个落地指南
  • 200+行重构前后对比代码
  • 可直接复用的3个装饰器工具类

项目背景与环境准备

betterpython是一个专注于Python代码质量提升的开源项目，包含了从基础代码整洁到高级架构设计的完整案例集合。项目基于YouTube系列教程"Write Better Python Code"构建，通过重构前后的代码对比，直观展示优秀Python代码的演进过程。

快速开始

# 克隆项目仓库
git clone https://gitcode.com/gh_mirrors/be/betterpython.git

# 进入项目目录
cd betterpython

# 安装依赖（如需要）
pip install -r requirements.txt

项目结构概览

betterpython/
├── 1 - coupling and cohesion/        # 耦合与内聚示例
├── 10 - object creation/             # 对象创建模式
├── 2 - dependency inversion/          # 依赖倒置原则
├── 3 - strategy pattern/             # 策略模式
├── 4 - observer pattern/             # 观察者模式
├── 5 - unit testing/                 # 单元测试实践
├── 6 - template method & bridge/     # 模板方法与桥接模式
├── 7 - dealing with errors/          # 错误处理机制
├── 8 - mvc/                          # MVC架构模式
└── 9 - solid/                        # SOLID原则完整案例

每个目录都包含before.py（重构前）和after.py（重构后）文件，直观展示代码改进过程。

代码整洁之道：从耦合到内聚

认识代码坏味道

在软件工程中，耦合（Coupling） 指模块之间的依赖程度，内聚（Cohesion） 指模块内部元素的关联程度。理想状态是低耦合高内聚（Loose Coupling, High Cohesion）。

以下是一个典型的高耦合低内聚代码示例：

# coupling-cohesion-before.py
import string
import random

def generate_vehicle_id(length):
    return ''.join(random.choices(string.ascii_uppercase, k=length))

def generate_vehicle_license(id):
    return f"{id[:2]}-{''.join(random.choices(string.digits, k=2))}-{id[-2:]}"

def register_vehicle(self, brand: string):
    # 一个函数做了太多事情！
    vehicle_id = generate_vehicle_id(12)
    license_plate = generate_vehicle_license(vehicle_id)
    
    # 直接处理数据库逻辑
    db = sqlite3.connect('vehicles.db')
    cursor = db.cursor()
    cursor.execute("INSERT INTO vehicles VALUES (?, ?)", (vehicle_id, license_plate))
    db.commit()
    db.close()
    
    return license_plate

问题分析：

  • 函数职责不单一：注册车辆同时处理ID生成、车牌生成和数据库操作
  • 硬编码依赖：直接依赖sqlite3，难以更换数据库
  • 缺乏封装：所有功能直接暴露为顶级函数

重构方案：职责分离与封装

重构后的代码通过引入专用类，实现了职责分离：

# coupling-cohesion-after.py
import string
import random

class VehicleInfo:
    """车辆信息类，专注于车辆属性和计算"""
    def __init__(self, brand, electric, catalogue_price):
        self.brand = brand
        self.electric = electric
        self.catalogue_price = catalogue_price
        
    def compute_tax(self):
        """计算车辆税费"""
        tax_percentage = 0.02 if self.electric else 0.05
        return tax_percentage * self.catalogue_price
        
    def print(self):
        """打印车辆信息"""
        print(f"Brand: {self.brand}")
        print(f"Payable tax: {self.compute_tax()}")


class Vehicle:
    """车辆类，封装车辆标识信息"""
    def __init__(self, id, license_plate, info):
        self.id = id
        self.license_plate = license_plate
        self.info = info
        
    def print(self):
        """打印车辆完整信息"""
        print(f"ID: {self.id}")
        print(f"License plate: {self.license_plate}")
        self.info.print()


class VehicleRegistry:
    """车辆注册类，专注于车辆注册流程"""
    def __init__(self):
        self.vehicles = []
        
    def add_vehicle_info(self, brand, electric, catalogue_price):
        """添加车辆信息模板"""
        self.vehicle_info = VehicleInfo(brand, electric, catalogue_price)
        
    def generate_vehicle_id(self, length):
        """生成车辆唯一ID"""
        return ''.join(random.choices(string.ascii_uppercase, k=length))
        
    def generate_vehicle_license(self, id):
        """生成车牌号码"""
        return f"{id[:2]}-{''.join(random.choices(string.digits, k=2))}-{id[-2:]}"
        
    def create_vehicle(self, brand):
        """创建新车辆"""
        vehicle_id = self.generate_vehicle_id(12)
        license_plate = self.generate_vehicle_license(vehicle_id)
        return Vehicle(vehicle_id, license_plate, self.vehicle_info)
        
    def register_vehicle(self, brand: string):
        """注册新车辆"""
        vehicle = self.create_vehicle(brand)
        self.vehicles.append(vehicle)
        return vehicle

改进点：

  • 职责分离：3个类分别处理车辆信息、车辆实体和注册流程
  • 封装加强：通过类方法隐藏实现细节
  • 可扩展性提升：添加新车辆类型只需扩展VehicleInfo

设计模式实战：从if-else到策略模式

场景引入：工单处理系统

假设我们需要实现一个工单处理系统，支持多种排序方式：FIFO（先进先出）、FILO（先进后出）、随机排序等。传统实现可能会使用大量条件判断：

# strategy-before.py
import string
import random
from typing import List

class SupportTicket:
    def __init__(self, customer, issue):
        self.id = generate_id()
        self.customer = customer
        self.issue = issue
        
    def generate_id(length=8):
        return ''.join(random.choices(string.ascii_uppercase, k=length))


class CustomerSupport:
    def __init__(self, processing_strategy: str = "fifo"):
        self.tickets = []
        self.processing_strategy = processing_strategy
        
    def create_ticket(self, customer, issue):
        self.tickets.append(SupportTicket(customer, issue))
        
    def process_tickets(self):
        # 这里有大量条件判断！
        if self.processing_strategy == "fifo":
            for ticket in self.tickets:
                self.process_ticket(ticket)
        elif self.processing_strategy == "filo":
            for ticket in reversed(self.tickets):
                self.process_ticket(ticket)
        elif self.processing_strategy == "random":
            for ticket in random.sample(self.tickets, len(self.tickets)):
                self.process_ticket(ticket)
        else:
            raise ValueError(f"Unknown strategy: {self.processing_strategy}")
            
    def process_ticket(self, ticket: SupportTicket):
        print(f"Processing ticket {ticket.id}: {ticket.customer} - {ticket.issue}")

问题分析：

  • 违反开闭原则：添加新排序方式需要修改process_tickets方法
  • 条件判断膨胀：每增加一种策略就增加一个elif分支
  • 代码复用困难：排序逻辑无法在其他系统中复用

策略模式重构方案

策略模式通过将算法封装为独立对象，实现了行为的动态选择：

# strategy-after.py
import string
import random
from typing import List
from abc import ABC, abstractmethod

class SupportTicket:
    def __init__(self, customer, issue):
        self.id = self.generate_id()
        self.customer = customer
        self.issue = issue
        
    def generate_id(self, length=8):
        return ''.join(random.choices(string.ascii_uppercase, k=length))


# 策略接口
class TicketOrderingStrategy(ABC):
    @abstractmethod
    def create_ordering(self, list: List[SupportTicket]) -> List[SupportTicket]:
        pass


# 具体策略实现
class FIFOOrderingStrategy(TicketOrderingStrategy):
    def create_ordering(self, list: List[SupportTicket]) -> List[SupportTicket]:
        return list.copy()


class FILOOrderingStrategy(TicketOrderingStrategy):
    def create_ordering(self, list: List[SupportTicket]) -> List[SupportTicket]:
        return list[::-1].copy()


class RandomOrderingStrategy(TicketOrderingStrategy):
    def create_ordering(self, list: List[SupportTicket]) -> List[SupportTicket]:
        return random.sample(list, len(list))


class CustomerSupport:
    def __init__(self, processing_strategy: TicketOrderingStrategy):
        self.tickets = []
        self.processing_strategy = processing_strategy  # 注入策略
        
    def create_ticket(self, customer, issue):
        self.tickets.append(SupportTicket(customer, issue))
        
    def process_tickets(self):
        # 委托给策略对象
        ordered_tickets = self.processing_strategy.create_ordering(self.tickets)
        
        for ticket in ordered_tickets:
            self.process_ticket(ticket)
            
    def process_ticket(self, ticket: SupportTicket):
        print(f"Processing ticket {ticket.id}: {ticket.customer} - {ticket.issue}")

Pythonic优化方案：

对于简单策略，Python可以直接使用函数作为策略，避免类的过度封装：

# strategy-after-fn.py
from dataclasses import dataclass, field
import string
import random
from typing import List, Callable

@dataclass
class SupportTicket:
    id: str = field(init=False)
    customer: str
    issue: str
    
    def __post_init__(self):
        self.id = self.generate_id()
        
    def generate_id(self, length=8):
        return ''.join(random.choices(string.ascii_uppercase, k=length))


# 策略函数
def fifo_ordering(list: List[SupportTicket]) -> List[SupportTicket]:
    return list.copy()


def filo_ordering(list: List[SupportTicket]) -> List[SupportTicket]:
    return list[::-1].copy()


def random_ordering(list: List[SupportTicket]) -> List[SupportTicket]:
    return random.sample(list, len(list))


@dataclass
class CustomerSupport:
    tickets: List[SupportTicket] = field(default_factory=list)
    # 使用Callable类型定义策略函数
    processing_strategy: Callable[[List[SupportTicket]], List[SupportTicket]] = fifo_ordering
        
    def create_ticket(self, customer, issue):
        self.tickets.append(SupportTicket(customer, issue))
        
    def process_tickets(self):
        # 直接调用策略函数
        ordered_tickets = self.processing_strategy(self.tickets)
        
        for ticket in ordered_tickets:
            self.process_ticket(ticket)
            
    def process_ticket(self, ticket: SupportTicket):
        print(f"Processing ticket {ticket.id}: {ticket.customer} - {ticket.issue}")


# 使用示例
if __name__ == "__main__":
    # 创建支持中心，指定策略
    support = CustomerSupport(processing_strategy=random_ordering)
    
    # 添加工单
    support.create_ticket("Alice", "无法登录")
    support.create_ticket("Bob", "支付失败")
    
    # 处理工单（将按随机顺序处理）
    support.process_tickets()

SOLID原则逐个突破

单一职责原则（Single Responsibility）

定义：一个类应该只有一个引起它变化的原因。

重构前问题代码：

# single-responsibility-before.py
class Order:
    def __init__(self):
        self.items = []
        self.status = "open"
        
    def add_item(self, name, quantity, price):
        self.items.append({"name": name, "quantity": quantity, "price": price})
        
    def total_price(self):
        return sum(item["price"] * item["quantity"] for item in self.items)
        
    def pay(self, payment_type, security_code):
        """一个方法处理多种支付类型，职责不单一"""
        if payment_type == "debit":
            print(f"Processing debit payment with security code {security_code}")
            self.status = "paid"
        elif payment_type == "credit":
            print(f"Processing credit payment with security code {security_code}")
            self.status = "paid"
        else:
            raise ValueError(f"Unknown payment type: {payment_type}")

重构后代码：

# single-responsibility-after.py
class Order:
    def __init__(self):
        self.items = []
        self.status = "open"
        
    def add_item(self, name, quantity, price):
        self.items.append({"name": name, "quantity": quantity, "price": price})
        
    def total_price(self):
        return sum(item["price"] * item["quantity"] for item in self.items)


class PaymentProcessor:
    """专注于支付处理的类"""
    def pay_debit(self, order, security_code):
        print(f"Processing debit payment with security code {security_code}")
        order.status = "paid"
        
    def pay_credit(self, order, security_code):
        print(f"Processing credit payment with security code {security_code}")
        order.status = "paid"

开放封闭原则（Open/Closed）

定义：软件实体应该对扩展开放，对修改关闭。

重构前问题代码：

# open-closed-before.py
class Order:
    def __init__(self):
        self.items = []
        self.status = "open"
        
    def add_item(self, name, quantity, price):
        self.items.append({"name": name, "quantity": quantity, "price": price})
        
    def total_price(self):
        return sum(item["price"] * item["quantity"] for item in self.items)


class PaymentProcessor:
    def pay(self, order, payment_type, security_code):
        """每增加一种支付方式都需要修改此方法"""
        if payment_type == "debit":
            print(f"Processing debit payment with security code {security_code}")
            order.status = "paid"
        elif payment_type == "credit":
            print(f"Processing credit payment with security code {security_code}")
            order.status = "paid"
        # 添加新支付方式需要修改这里
        else:
            raise ValueError(f"Unknown payment type: {payment_type}")

重构后代码：

# open-closed-after.py
from abc import ABC, abstractmethod

class Order:
    def __init__(self):
        self.items = []
        self.status = "open"
        
    def add_item(self, name, quantity, price):
        self.items.append({"name": name, "quantity": quantity, "price": price})
        
    def total_price(self):
        return sum(item["price"] * item["quantity"] for item in self.items)


class PaymentProcessor(ABC):
    """支付处理器抽象基类"""
    @abstractmethod
    def pay(self, order, security_code):
        pass


class DebitPaymentProcessor(PaymentProcessor):
    """借记卡支付处理器"""
    def pay(self, order, security_code):
        print(f"Processing debit payment with security code {security_code}")
        order.status = "paid"


class CreditPaymentProcessor(PaymentProcessor):
    """信用卡支付处理器"""
    def pay(self, order, security_code):
        print(f"Processing credit payment with security code {security_code}")
        order.status = "paid"


class PayPalPaymentProcessor(PaymentProcessor):
    """PayPal支付处理器（新扩展，无需修改现有代码）"""
    def pay(self, order, security_code):
        print(f"Processing PayPal payment with email {security_code}")
        order.status = "paid"

其他SOLID原则实现

由于篇幅限制，我们简要介绍剩余三个原则在项目中的实现：

里氏替换原则（Liskov Substitution）

核心思想：子类应该可以替换父类并出现在父类能够出现的任何地方。

项目中的实现案例：

  • 9 - solid/liskov-substitution-after.py
  • 通过统一接口确保子类与父类的可替换性
  • 避免子类抛出父类未声明的异常

接口隔离原则（Interface Segregation）

核心思想：客户端不应该依赖它不需要的接口。

项目中的实现案例：

  • 9 - solid/interface-segregation-after.py
  • 将大接口拆分为多个小接口
  • 如将PaymentProcessor拆分为PaymentMethod和SMSAuthorizer

依赖倒置原则（Dependency Inversion）

核心思想：依赖抽象而不是具体实现。

项目中的实现案例：

  • 2 - dependency inversion/dependency-inversion-after.py
  • 通过构造函数注入依赖
  • 高层模块不依赖低层模块，二者都依赖抽象
# dependency-inversion-after.py
from abc import ABC, abstractmethod

class Order:
    def __init__(self):
        self.items = []
        self.status = "open"
        
    def add_item(self, name, quantity, price):
        self.items.append({"name": name, "quantity": quantity, "price": price})
        
    def total_price(self):
        return sum(item["price"] * item["quantity"] for item in self.items)


class Authorizer(ABC):
    """授权器抽象基类"""
    @abstractmethod
    def is_authorized(self) -> bool:
        pass


class SMSAuthorizer(Authorizer):
    """短信授权器"""
    def __init__(self):
        self.authorized = False
        
    def verify_code(self, code):
        print(f"Verifying SMS code {code}")
        self.authorized = True
        
    def is_authorized(self) -> bool:
        return self.authorized


class PaymentProcessor(ABC):
    """支付处理器抽象基类"""
    def __init__(self, authorizer: Authorizer):
        self.authorizer = authorizer  # 依赖抽象而非具体实现
        
    @abstractmethod
    def pay(self, order):
        pass


class DebitPaymentProcessor(PaymentProcessor):
    def __init__(self, security_code, authorizer: Authorizer):
        super().__init__(authorizer)
        self.security_code = security_code
        
    def pay(self, order):
        if not self.authorizer.is_authorized():
            raise Exception("Not authorized")
        print(f"Processing debit payment with security code {self.security_code}")
        order.status = "paid"

实用工具与高级技巧

错误处理机制

项目7 - dealing with errors/目录提供了多种错误处理方案，从基础到高级依次为：

  1. 原始错误处理：before/error-handling.py

    • 使用简单的try-except块
    • 缺乏统一错误处理机制

  2. 集中式错误处理：after/error-handling.py

    • 自定义异常类型
    • 集中式错误处理逻辑

  3. 上下文管理器：after-context/error-handling-context.py

    • 使用with语句管理资源
    • 自动处理异常和资源释放

  4. 函数式错误处理：monadic-error-handling/example.py

    • 使用returns库实现Result类型
    • 函数组合处理错误流程
# monadic-error-handling/example.py
from returns.result import Result, safe
from returns.pipeline import flow
from returns.pointfree import bind

class Blog:
    def __init__(self, id, title):
        self.id = id
        self.title = title

def fetch_blog_from_db(blog_id):
    # 模拟数据库查询
    if blog_id == 1:
        return Blog(1, "Python设计模式")
    raise ValueError("Blog not found")

@safe
def blog_to_dict(item) -> Blog:
    return {"id": item.id, "title": item.title}

@safe
def verify_access(blog) -> Blog:
    # 模拟权限验证
    return blog

def fetch_blog(blog_id) -> Result[Blog, Exception]:
    """使用函数式管道处理错误"""
    return flow(
        blog_id,
        fetch_blog_from_db,  # 可能抛出异常
        bind(verify_access),  # 权限验证
        bind(blog_to_dict)    # 转换为字典
    )

# 使用示例
result = fetch_blog(1)
if result.is_success():
    print(result.unwrap())  # 成功时获取结果
else:
    print(f"Error: {result.failure()}")  # 失败时处理错误

实用装饰器

项目7 - dealing with errors/advanced/提供了两个非常实用的装饰器：

1. 日志装饰器

# logging-decorator.py
import logging 
from functools import wraps

def create_logger():
    logger = logging.getLogger("example")
    logger.setLevel(logging.INFO)
    return logger

def exception(logger):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.exception(f"Exception in {func.__name__}: {e}")
                raise  # 可以选择是否重新抛出异常
        return wrapper
    return decorator

# 使用示例
logger = create_logger()

@exception(logger)
def divideByZero():
    return 1 / 0

divideByZero()

2. 重试装饰器

# retry-decorator.py
import time
import math
from functools import wraps

def retry(ExceptionToCheck, tries=4, delay=3, backoff=2, logger=None):
    """
    带退避策略的重试装饰器
    
    :param ExceptionToCheck: 需要捕获的异常类型
    :param tries: 最大重试次数
    :param delay: 初始延迟（秒）
    :param backoff: 退避乘数，如2表示每次延迟翻倍
    """
    def deco_retry(f):
        @wraps(f)
        def f_retry(*args, **kwargs):
            mtries, mdelay = tries, delay
            while mtries > 1:
                try:
                    return f(*args, **kwargs)
                except ExceptionToCheck as e:
                    if logger:
                        logger.warning(f"{str(e)}, 剩余重试次数: {mtries - 1}")
                    time.sleep(mdelay)
                    mtries -= 1
                    mdelay *= backoff
            return f(*args, **kwargs)
        return f_retry
    return deco_retry

# 使用示例
@retry(ValueError, tries=3, delay=2)
def test_fail(text):
    if text != "success":
        raise ValueError("测试失败")
    return "成功"

总结与后续学习

通过betterpython项目的案例学习，我们掌握了从代码整洁到架构设计的完整路径。关键收获包括：

  1. 代码质量：通过耦合与内聚分析，识别并改进代码坏味道
  2. 设计模式：策略模式、观察者模式等在Python中的最佳实践
  3. 架构原则：SOLID原则的逐个落地方法
  4. 错误处理：从简单try-except到函数式错误处理的演进
  5. Pythonic技巧：使用函数作为策略、装饰器等Python特色功能

项目最佳实践清单

  • 代码组织：按功能或领域划分模块，而非按模式或层次
  • 命名规范：使用清晰的动词+名词命名方式，如fetch_blog而非get_blog
  • 依赖管理：通过构造函数注入依赖，提高可测试性
  • 错误处理：定义项目统一的错误处理策略，避免到处try-except
  • 测试策略：先重构为可测试代码，再编写单元测试

推荐后续学习路径

  1. 深入设计模式：10 - object creation/目录的单例模式和对象池模式
  2. MVC架构：8 - mvc/目录的MVC模式实现
  3. 观察者模式：4 - observer pattern/目录的事件驱动设计
  4. 单元测试：5 - unit testing/目录的测试实践

希望本文能帮助你写出更优雅、更可维护的Python代码。如果你觉得本文有价值，请点赞、收藏并关注作者，下期将带来"Python元编程高级技巧"专题。

betterpython
Code examples for my Write Better Python Code series on YouTube.
项目地址：https://gitcode.com/gh_mirrors/be/betterpython
登录后查看全文
热门项目推荐
相关项目推荐

项目优选

收起
kernelkernel
deepin linux kernel
C
27
11
docsdocs
OpenHarmony documentation | OpenHarmony开发者文档
Dockerfile
478
3.57 K
ohos_react_nativeohos_react_native
React Native鸿蒙化仓库
JavaScript
287
340
flutter_flutterflutter_flutter
暂无简介
Dart
728
175
pytorchpytorch
Ascend Extension for PyTorch
Python
288
321
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库，实现网络在NPU上加速计算。
C++
850
447
kernelkernel
openEuler内核是openEuler操作系统的核心，既是系统性能与稳定性的基石，也是连接处理器、设备与服务的桥梁。
C
239
100
nop-entropynop-entropy
Nop Platform 2.0是基于可逆计算理论实现的采用面向语言编程范式的新一代低代码开发平台，包含基于全新原理从零开始研发的GraphQL引擎、ORM引擎、工作流引擎、报表引擎、规则引擎、批处理引引擎等完整设计。nop-entropy是它的后端部分，采用java语言实现，可选择集成Spring框架或者Quarkus框架。中小企业可以免费商用
Java
10
1
torchairtorchair
TorchAir 支持用户基于PyTorch框架和torch_npu插件在昇腾NPU上使用图模式进行推理。
Python
451
180
RuoYi-Vue3RuoYi-Vue3
🎉 (RuoYi)官方仓库 基于SpringBoot，Spring Security，JWT，Vue3 & Vite、Element Plus 的前后端分离权限管理系统
Vue
1.28 K
705