编程进阶网 编程进阶网
首页
  • 计算机原理
  • 操作系统
  • 网络协议
  • 数据库原理
  • 面向对象
  • 设计原则
  • 设计模式
  • 系统架构
  • 性能优化
  • 编程原理
  • 方案设计
  • 稳定可靠
  • 工程运维
  • 基础认知
  • 线性结构
  • 树与哈希
  • 工业级实现
  • 算法思想
  • 实战与综合
  • 算法题考核
  • C语言入门
  • C综合案例
  • C专栏博客
  • C标准集库
  • C++入门教程
  • C++综合案例
  • C++专栏博客
  • C++开发技巧
  • Java入门教程
  • Java综合案例
  • Java专栏博客
  • Go入门教程
  • Go综合案例
  • Go专栏博客
  • Go开发技巧
  • JavaScript入门
  • JavaScript高级
  • Android库解读
  • Android专栏
  • Android智能硬件
  • iOS ObjC入门
  • iOS Swift入门
  • iOS入门精通
  • Web之Html手册
  • Web之TypeScript
  • Web之Vue高级进阶
  • Linux之QML入门
  • Linux之QT核心库
  • Linux实践开发
  • Python教程
  • Shell&Bash教程
  • 工具脚本
  • 自动化脚本
  • 质量保障
  • 产品思考
  • 软实力
  • 开发流程
  • Git应用
  • 技术模版
  • 技术规范
  • Markdown
  • Mermaid
  • 开源协议
  • JSON工具
  • 文本工具
  • 图片处理
  • 文档转化
  • 代码压缩
  • 关于我
  • 自我精进
  • 职场管理
  • 职场面试
  • 心情杂货
  • 友情链接

杨充

专注编程 · 终身学习者
首页
  • 计算机原理
  • 操作系统
  • 网络协议
  • 数据库原理
  • 面向对象
  • 设计原则
  • 设计模式
  • 系统架构
  • 性能优化
  • 编程原理
  • 方案设计
  • 稳定可靠
  • 工程运维
  • 基础认知
  • 线性结构
  • 树与哈希
  • 工业级实现
  • 算法思想
  • 实战与综合
  • 算法题考核
  • C语言入门
  • C综合案例
  • C专栏博客
  • C标准集库
  • C++入门教程
  • C++综合案例
  • C++专栏博客
  • C++开发技巧
  • Java入门教程
  • Java综合案例
  • Java专栏博客
  • Go入门教程
  • Go综合案例
  • Go专栏博客
  • Go开发技巧
  • JavaScript入门
  • JavaScript高级
  • Android库解读
  • Android专栏
  • Android智能硬件
  • iOS ObjC入门
  • iOS Swift入门
  • iOS入门精通
  • Web之Html手册
  • Web之TypeScript
  • Web之Vue高级进阶
  • Linux之QML入门
  • Linux之QT核心库
  • Linux实践开发
  • Python教程
  • Shell&Bash教程
  • 工具脚本
  • 自动化脚本
  • 质量保障
  • 产品思考
  • 软实力
  • 开发流程
  • Git应用
  • 技术模版
  • 技术规范
  • Markdown
  • Mermaid
  • 开源协议
  • JSON工具
  • 文本工具
  • 图片处理
  • 文档转化
  • 代码压缩
  • 关于我
  • 自我精进
  • 职场管理
  • 职场面试
  • 心情杂货
  • 友情链接
  • ScriptHub 脚本工具箱
  • Python

    • Python 从入门到实战
    • 入门与基础类型
    • 序列与集合类型
    • 流程控制与函数
    • 面向对象与工程
      • 4.1 面向对象编程
        • 4.1.1 class 与对象
        • 4.1.2 继承与多态
        • 4.1.3 私有属性与封装
        • 4.1.4 @property 属性访问器
        • 4.1.5 特殊方法(魔术方法)
        • 4.1.6 综合案例与思考
      • 4.2 模块与包
        • 4.2.1 import 的四种方式
        • 4.2.2 __name__ == '__main__'
        • 4.2.3 包结构与 __init__.py
        • 4.2.4 pip 与虚拟环境
        • 4.2.5 综合案例与思考
      • 4.3 文件与异常
        • 4.3.1 文件读写操作
        • 4.3.2 with 上下文管理器
        • 4.3.3 json 与 csv 数据格式
        • 4.3.4 try-except-else-finally
        • 4.3.5 自定义异常类
        • 4.3.6 综合案例与思考
      • 4.4 迭代器与生成器
        • 4.4.1 iter/next 协议
        • 4.4.2 yield 生成器函数
        • 4.4.3 生成器表达式
        • 4.4.4 itertools 常用工具
        • 4.4.5 生成器节省内存原理
        • 4.4.6 综合案例与思考
      • 4.5 新手陷阱
      • 4.6 综合思考题
      • 4.7 Python 基础入门知识地图
    • 爬虫全流程实战
    • 数据分析三件套
    • 办公自动化实战
    • 开发环境与规范
    • 调试与性能优化
    • 部署与并发实战
    • 函数高级特性剖析:装饰器 / 生成器 / 上下文管理器
    • 并发底层原理揭秘
    • 面向对象与类型系统:元类 / 描述符 / 鸭子类型
    • 解释器源码初探
  • Shell-Bash

  • 工具脚本

  • ScriptHub
  • Python
杨充
2019-06-29
目录

面向对象与工程

# 第 4 章 Python 面向对象与工程

# 目录介绍

  • 4.1 面向对象编程
    • 4.1.1 class 与对象
    • 4.1.2 继承与多态
    • 4.1.3 私有属性与封装
    • 4.1.4 @property 属性访问器
    • 4.1.5 特殊方法(魔术方法)
    • 4.1.6 综合案例与思考
  • 4.2 模块与包
    • 4.2.1 import 的四种方式
    • 4.2.2 __name__ == '__main__'
    • 4.2.3 包结构与 __init__.py
    • 4.2.4 pip 与虚拟环境
    • 4.2.5 综合案例与思考
  • 4.3 文件与异常
    • 4.3.1 文件读写操作
    • 4.3.2 with 上下文管理器
    • 4.3.3 json 与 csv 数据格式
    • 4.3.4 try-except-else-finally
    • 4.3.5 自定义异常类
    • 4.3.6 综合案例与思考
  • 4.4 迭代器与生成器
    • 4.4.1 iter/next 协议
    • 4.4.2 yield 生成器函数
    • 4.4.3 生成器表达式
    • 4.4.4 itertools 常用工具
    • 4.4.5 生成器节省内存原理
    • 4.4.6 综合案例与思考
  • 4.5 新手陷阱 Top 5
  • 4.6 综合思考题

# 4.1 面向对象编程

📖 本章定位:Python 基础入门最后一章——OOP + 模块化 + 文件 IO + 异常 + 迭代器/生成器。学完后你具备了"工程化 Python"的能力。

📌 前章回顾:第 3 章的函数、闭包、装饰器是本章的基础——class 的方法就是特殊的函数,@property 就是函数装饰器。如果你对 LEGB 作用域规则还模糊,请先回顾 §3.4。 📌 知识地图:本章末尾 §4.7 有四章知识全景图——建议学完全章后回看。

Python 的 OOP 和 C++/Java 有两大根本不同:① 没有 private 关键字——全靠约定 + 名称改写;② 一切皆对象——类本身也是对象。

# 4.1.1 class 与对象

class Student:
    """学生类——展示类的基本定义"""

    # 类属性(所有实例共享——相当于 C++ 的 static 成员)
    school = "深圳大学"

    def __init__(self, name: str, age: int, score: float = 0.0):
        """构造方法——创建对象时自动调用"""
        self.name = name         # 实例属性——每个对象独立
        self.age = age
        self._score = score      # _ 开头表示"受保护"(约定)

    def introduce(self):
        """实例方法——第一个参数必须是 self"""
        return f"我叫{self.name},{self.age}岁,就读于{self.school}"

    def get_grade(self) -> str:
        if self._score >= 90:   return "A"
        elif self._score >= 80: return "B"
        elif self._score >= 70: return "C"
        return "D"


# 创建对象
s1 = Student("张三", 20, 85)
s2 = Student("李四", 21, 92)

print(s1.introduce())     # 我叫张三,20岁,就读于深圳大学
print(s2.get_grade())     # A

# 类属性共享
print(s1.school)          # 深圳大学
print(s2.school)          # 深圳大学
Student.school = "北京大学"  # 改了类属性——所有实例都受影响
print(s1.school)          # 北京大学
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

🔑 self 不是关键字——只是约定俗成的名字:

class Demo:
    def method(this):     # 可以叫 this,但没人这么干
        print(this)

# 等价写法:类名.方法(实例)
s1 = Student("张三", 20)
print(Student.introduce(s1))  # 和 s1.introduce() 完全等价
1
2
3
4
5
6
7

# 4.1.2 继承与多态

class Animal:
    def __init__(self, name):
        self.name = name

    def speak(self):
        """子类应该重写这个方法"""
        raise NotImplementedError("子类必须实现 speak()")


class Dog(Animal):
    def speak(self):
        return f"{self.name}:汪汪!"

    def wag_tail(self):
        return f"{self.name} 摇了摇尾巴"


class Cat(Animal):
    def speak(self):
        return f"{self.name}:喵喵~"


# 多态:同一接口、不同行为
animals = [Dog("旺财"), Cat("咪咪"), Dog("大黄")]
for a in animals:
    print(a.speak())
# 旺财:汪汪!
# 咪咪:喵喵~
# 大黄:汪汪!

# 检查类型
d = Dog("小黑")
print(isinstance(d, Dog))       # True
print(isinstance(d, Animal))    # True(子类也是父类)
print(issubclass(Dog, Animal))  # True

# super():调用父类方法
class Puppy(Dog):
    def __init__(self, name, toy):
        super().__init__(name)  # 调用 Dog 的 __init__(最终到 Animal)
        self.toy = toy

    def speak(self):
        return super().speak() + f" 我喜欢玩{self.toy}!"

p = Puppy("小黄", "骨头")
print(p.speak())              # 小黄:汪汪! 我喜欢玩骨头!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

🔑 Python 支持多重继承——用 MRO(方法解析顺序)确定查找路径:

class A:
    def say(self): return "A"

class B(A):
    def say(self): return "B"

class C(A):
    def say(self): return "C"

class D(B, C):        # 同时继承 B 和 C——多重继承
    pass

d = D()
print(d.say())         # B(从左往右查找——MRO 规则)
print(D.__mro__)       # D → B → C → A → object
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

⚠️ 多重继承很危险——菱形继承会导致"某个父类的 __init__ 被调用两次"。生产代码优先用组合而非多重继承。

# 4.1.3 私有属性与封装

Python 没有真正的 private——靠约定和名称改写:

class BankAccount:
    def __init__(self, owner: str, balance: float):
        self.owner = owner                      # 公开——可直接访问
        self._balance = balance                 # _ 保护——"请别碰"(约定)
        self.__pin = "1234"                     # __ 私有——名称改写

    def deposit(self, amount: float):
        self._balance += amount
        return self._balance

    def withdraw(self, amount: float, pin: str) -> float:
        if pin != self.__pin:
            raise PermissionError("PIN 错误")
        if amount > self._balance:
            raise ValueError("余额不足")
        self._balance -= amount
        return amount

    def get_balance(self):
        return self._balance


acc = BankAccount("张三", 1000)

# 公开属性——直接访问
print(acc.owner)               # 张三

# _ 保护属性——可以访问但不应访问
print(acc._balance)            # 1000(PEP 8 说"你不该这么做")

# __ 私有属性——名称被改写为 _BankAccount__pin
# print(acc.__pin)             # AttributeError! 找不到
print(acc._BankAccount__pin)   # '1234'(被改写后仍能访问——Python 哲学:我们都是成年人)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

🔑 名称改写规则:__xxx → _ClassName__xxx——这是为了防止子类意外覆盖,不是安全机制。

# 4.1.4 @property 属性访问器

@property 让方法像属性一样访问——Python OOP 最优雅的特性之一:

class Temperature:
    def __init__(self, celsius: float):
        self._celsius = celsius

    @property
    def celsius(self) -> float:
        """获取摄氏温度"""
        return self._celsius

    @celsius.setter
    def celsius(self, value: float):
        """设置摄氏温度——带校验"""
        if value < -273.15:
            raise ValueError("不能低于绝对零度!")
        self._celsius = value

    @property
    def fahrenheit(self) -> float:
        """华氏温度——只读属性(没有 setter)"""
        return self._celsius * 9 / 5 + 32

    @property
    def kelvin(self) -> float:
        """开尔文温度——只读"""
        return self._celsius + 273.15


t = Temperature(25)
print(t.celsius)        # 25(像属性一样读——实际调用了 getter)
print(t.fahrenheit)     # 77.0
print(t.kelvin)         # 298.15

t.celsius = 100         # 像属性一样写——实际调用了 setter,有校验
print(t.celsius)        # 100

# t.celsius = -500      # ValueError:不能低于绝对零度!
# t.fahrenheit = 80     # AttributeError:没有 setter
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

🔑 对比 C++:

// C++:需要显式 getter/setter
class Temperature {
    double celsius_;
public:
    double celsius() const { return celsius_; }
    void set_celsius(double v) { celsius_ = v; }
};
// 调用:t.celsius();     t.set_celsius(100);   ← 方法调用的语法,暴露了"是方法"
1
2
3
4
5
6
7
8
# Python:@property 封装后,语法和属性一样
t.celsius          # 读——看不出是方法
t.celsius = 100    # 写——看不出是方法
# 这就是 Python 的"统一访问原则"——接口稳定不变,内部可以随时改成属性或方法
1
2
3
4

# 4.1.5 特殊方法(魔术方法)

双下划线方法(dunder methods)让自定义类融入 Python 的内置语法:

class Vector:
    def __init__(self, x: float, y: float):
        self.x, self.y = x, y

    # ----- 字符串表示 -----
    def __repr__(self):
        """给开发者看的——对 eval() 友好"""
        return f"Vector({self.x}, {self.y})"

    def __str__(self):
        """给用户看的——print() 调用"""
        return f"({self.x}, {self.y})"

    # ----- 数学运算 -----
    def __add__(self, other):
        return Vector(self.x + other.x, self.y + other.y)

    def __sub__(self, other):
        return Vector(self.x - other.x, self.y - other.y)

    def __mul__(self, scalar: float):
        return Vector(self.x * scalar, self.y * scalar)

    def __neg__(self):
        return Vector(-self.x, -self.y)

    def __abs__(self):
        return (self.x ** 2 + self.y ** 2) ** 0.5

    # ----- 比较 -----
    def __eq__(self, other):
        return self.x == other.x and self.y == other.y

    def __lt__(self, other):
        return abs(self) < abs(other)

    # ----- 容器行为 -----
    def __len__(self):
        return 2   # 当作二维向量——"有两个分量"

    def __getitem__(self, index):
        if index == 0: return self.x
        if index == 1: return self.y
        raise IndexError("索引只能 0 或 1")

    # ----- 可调用 -----
    def __call__(self):
        """让实例像函数一样被调用"""
        return self.x ** 2 + self.y ** 2


# 使用——不需要学新 API,全是 Python 原生语法
v1 = Vector(3, 4)
v2 = Vector(1, 2)

print(v1 + v2)              # (4, 6) ← __add__
print(v1 - v2)              # (2, 2) ← __sub__
print(v1 * 3)               # (9, 12) ← __mul__
print(-v1)                  # (-3, -4) ← __neg__
print(abs(v1))              # 5.0 ← __abs__
print(v1 == Vector(3, 4))   # True ← __eq__
print(v1 > v2)              # True ← __lt__
print(len(v1))              # 2 ← __len__
print(v1[0], v1[1])         # 3 4 ← __getitem__
print(v1())                 # 25 ← __call__(3² + 4²)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

常用特殊方法速查表:

方法 触发条件 类比
__init__ obj = Class() 构造函数
__str__ print(obj), str(obj) toString
__repr__ 交互模式直接输入 obj 调试表示
__add__ a + b 运算符重载
__eq__ a == b 相等比较
__len__ len(obj) 长度
__getitem__ obj[i] 索引访问
__iter__ for x in obj 迭代器
__enter__/__exit__ with obj: 上下文管理器
__call__ obj() 可调用对象

# 4.1.6 综合案例与思考

综合案例:简易银行系统——OOP 全家桶

"""
银行系统——class/继承/封装/@property/特殊方法的综合运用
"""

from datetime import datetime
from typing import Optional


class Transaction:
    """交易记录"""
    def __init__(self, amount: float, txn_type: str):
        self.time = datetime.now()
        self.amount = amount
        self.type = txn_type  # "deposit" / "withdraw" / "transfer_in" / "transfer_out"

    def __str__(self):
        sign = "+" if self.type in ("deposit", "transfer_in") else "-"
        return f"{self.time.strftime('%m-%d %H:%M')}  {sign}¥{self.amount:,.2f}  [{self.type}]"


class Account:
    """银行账户——封装 + @property + 特殊方法"""
    _id_counter = 0

    def __init__(self, owner: str, balance: float = 0.0):
        Account._id_counter += 1
        self._id = Account._id_counter
        self.owner = owner
        self._balance = balance
        self._transactions: list[Transaction] = []
        self.is_active = True

    # ----- @property 访问器 -----
    @property
    def id(self) -> int:
        return self._id

    @property
    def balance(self) -> float:
        return self._balance

    @balance.setter
    def balance(self, value):
        if value < 0:
            raise ValueError("余额不能为负数")
        self._balance = value

    @property
    def transaction_count(self) -> int:
        return len(self._transactions)

    # ----- 核心操作 -----
    def deposit(self, amount: float) -> bool:
        if amount <= 0:
            print("存款金额必须为正")
            return False
        self._balance += amount
        self._transactions.append(Transaction(amount, "deposit"))
        return True

    def withdraw(self, amount: float) -> bool:
        if amount <= 0:
            print("取款金额必须为正")
            return False
        if amount > self._balance:
            print(f"余额不足——当前余额 ¥{self._balance:,.2f}")
            return False
        self._balance -= amount
        self._transactions.append(Transaction(amount, "withdraw"))
        return True

    def transfer_to(self, target: "Account", amount: float) -> bool:
        if self.withdraw(amount):
            target._balance += amount
            target._transactions.append(Transaction(amount, "transfer_in"))
            self._transactions[-1].type = "transfer_out"  # 覆盖类型
            return True
        return False

    # ----- 特殊方法 -----
    def __str__(self):
        status = "🟢" if self.is_active else "🔴"
        return f"{status} [{self._id:04d}] {self.owner}  ¥{self._balance:,.2f}"

    def __repr__(self):
        return f"Account(owner='{self.owner}', balance={self._balance})"

    def __eq__(self, other):
        return self._id == other._id

    def __lt__(self, other):
        return self._balance < other._balance

    def __len__(self):
        return self.transaction_count

    # ----- 历史打印 -----
    def show_history(self, limit: int = 10):
        print(f"\n--- {self.owner} 的交易记录(最近 {min(limit, len(self))} 笔)---")
        for t in self._transactions[-limit:]:
            print(f"  {t}")


class SavingsAccount(Account):
    """储蓄账户——继承 + 多态"""
    RATE = 0.025  # 年利率 2.5%

    def __init__(self, owner: str, balance: float = 0.0):
        super().__init__(owner, balance)
        self._interest_earned = 0.0

    def apply_interest(self):
        interest = self._balance * self.RATE / 12  # 月利息
        self._interest_earned += interest
        self._balance += interest
        self._transactions.append(Transaction(interest, "interest"))

    # 多态——覆盖父类方法
    def __str__(self):
        base = super().__str__()
        return f"{base}  💰{self._interest_earned:,.2f}利息"


class Bank:
    """银行——管理所有账户"""
    def __init__(self, name: str):
        self.name = name
        self.accounts: dict[int, Account] = {}

    def create_account(self, owner: str, initial: float = 0.0, savings: bool = False) -> Account:
        cls = SavingsAccount if savings else Account
        acc = cls(owner, initial)
        self.accounts[acc.id] = acc
        return acc

    def get_account(self, acc_id: int) -> Optional[Account]:
        return self.accounts.get(acc_id)

    def total_assets(self) -> float:
        return sum(a.balance for a in self.accounts.values())

    def __len__(self):
        return len(self.accounts)

    def __str__(self):
        lines = [f"🏦 {self.name}({len(self)} 个账户 | 总资产 ¥{self.total_assets():,.2f})"]
        for acc in sorted(self.accounts.values(), reverse=True):   # 按余额排序
            lines.append(f"  {acc}")
        return "\n".join(lines)


# ===== 运行演示 =====
def demo():
    bank = Bank("Python 银行")

    # 开户
    a1 = bank.create_account("张三", 10000, savings=True)
    a2 = bank.create_account("李四", 50000)
    a3 = bank.create_account("王五", 3000)

    # 交易
    a1.deposit(5000)
    a2.withdraw(10000)
    a2.transfer_to(a3, 20000)

    # 储蓄账户计息
    if isinstance(a1, SavingsAccount):
        a1.apply_interest()

    # 查看
    print(bank)
    print(f"\n{'='*50}")
    a2.show_history(5)
    a1.show_history(5)
    a3.show_history(5)

    print(f"\n{'='*50}")
    print(f"李四 > 王五? {a2 > a3}(余额比较)")
    print(f"张三交易次数:{len(a1)} 笔")

demo()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

运行片段:

🏦 Python 银行(3 个账户 | 总资产 ¥55,031.25)
  🟢 [0002] 李四  ¥20,000.00
  🟢 [0001] 张三  ¥15,031.25  💰31.25利息
  🟢 [0003] 王五  ¥20,000.00

==================================================
--- 李四 的交易记录(最近 3 笔)---
  06-08 10:00  -¥10,000.00  [withdraw]
  06-08 10:00  -¥20,000.00  [transfer_out]

--- 张三 的交易记录(最近 2 笔)---
  06-08 10:00  +¥5,000.00  [deposit]
  06-08 10:00  +¥31.25  [interest]
==================================================
李四 > 王五? False(余额比较)
张三交易次数:2 笔
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

案例知识融合:这个案例在真实场景中综合运用了 OOP 全部核心——__init__ 构造、类属性(_id_counter)、@property 封装、继承(SavingsAccount)、多态(重写 __str__)、特殊方法(__str__/__repr__/__eq__/__lt__/__len__)、组合(Bank 包含 Account)、isinstance 类型判断——120 行完成了一个可用的银行系统。

思考题:

  1. __lt__ 实现余额比较——如果两个账户余额相同,a1 < a2 和 a1 <= a2 分别返回什么?Python 的排序是稳定的吗?
  2. self._transactions[-1].type = "transfer_out" 直接修改最后一个交易的类型——这样的设计有什么问题?如果用不可变对象(namedtuple)替代怎么办?
  3. __repr__ 和 __str__ 的设计原则是什么?为什么 Vector 案例里 __repr__ 返回 Vector(3, 4) 而不是 (3, 4)?

# 4.2 模块与包

# 4.2.1 import 的四种方式

# 方式一:导入整个模块
import math
print(math.sqrt(16))         # 4.0

# 方式二:导入特定函数/类
from math import sqrt, pi
print(sqrt(25))              # 5.0
print(pi)                    # 3.141592...

# 方式三:导入并重命名(解决命名冲突)
import numpy as np
from datetime import datetime as dt

# 方式四:导入全部(不推荐——污染命名空间)
# from math import *         # ❌ 大量符号涌入,看不出来源
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

🔑 import 的搜索顺序:当前目录 → PYTHONPATH 环境变量 → 标准库 → 第三方包(sys.path)。

# 4.2.2 __name__ == '__main__'

这是 Python 的"入口守卫"——让同一个文件既能作为脚本运行、又能作为模块被导入:

# my_utils.py
def add(a, b):
    return a + b

def subtract(a, b):
    return a - b

# 只有当直接运行 python my_utils.py 时才执行
if __name__ == "__main__":
    print("测试 add:", add(3, 5))        # 测试 add: 8
    print("测试 subtract:", subtract(10, 3))  # 测试 subtract: 7
1
2
3
4
5
6
7
8
9
10
11
# main.py——导入 my_utils
import my_utils

print(my_utils.add(1, 2))    # 3——不会触发 my_utils 的测试代码
1
2
3
4

🔑 原理:当用 python my_utils.py 运行时,__name__ 被设为 "__main__";当被 import my_utils 时,__name__ 是 "my_utils"。

# 4.2.3 包结构与 __init__.py

包 = 含 __init__.py 的目录——组织代码的层级单元:

my_project/
├── my_project/
│   ├── __init__.py           # 标记为包;可空,也可写导入逻辑
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py           # class User
│   │   └── product.py        # class Product
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── validators.py
│   │   └── helpers.py
│   └── main.py
├── tests/
│   └── test_user.py
└── setup.py                  # 打包配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

__init__.py 的三种用法:

# 1. 空文件——仅标记这是包
# my_project/models/__init__.py(空)

# 2. 控制导出接口
from .user import User
from .product import Product
__all__ = ["User", "Product"]   # 限制 from models import * 暴露什么

# 3. 包级别的初始化
import logging
logging.getLogger(__name__).addHandler(logging.NullHandler())
1
2
3
4
5
6
7
8
9
10
11

🔑 相对导入:包内部用 . 表示层级——只在包内部有效:

# my_project/models/user.py——想引用同包下的 product
from .product import Product        # . 表示当前包
from ..utils.validators import validate_email  # .. 表示上级包
1
2
3

# 4.2.4 pip 与虚拟环境

# ===== pip:Python 的包管理器 =====
pip install requests              # 安装
pip install requests==2.31.0      # 指定版本
pip install "django>=4.0,<5.0"    # 版本范围
pip uninstall requests            # 卸载
pip list                          # 列出已安装
pip freeze > requirements.txt     # 导出依赖清单
pip install -r requirements.txt   # 恢复依赖

# ===== 虚拟环境:项目隔离 =====
python -m venv myenv              # 创建虚拟环境
source myenv/bin/activate         # 激活(Linux/macOS)
myenv\Scripts\activate            # 激活(Windows)
deactivate                        # 退出

# 激活后,pip install 只影响当前项目——不再污染全局 Python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

⚠️ 永远不要 pip install 到全局 Python。每个项目一个虚拟环境——这是 Python 工程化的第一准则。

# 4.2.5 综合案例与思考

综合案例:学生管理系统——多模块项目结构

student_system/
├── main.py                       # 入口
├── models/
│   ├── __init__.py               # 导出 Student, Course
│   ├── student.py
│   └── course.py
├── services/
│   ├── __init__.py
│   └── enrollment.py             # 选课服务
└── utils/
    ├── __init__.py
    └── helpers.py
1
2
3
4
5
6
7
8
9
10
11
12
# ===== models/student.py =====
class Student:
    def __init__(self, sid: int, name: str):
        self.sid = sid
        self.name = name
        self.courses: list["Course"] = []   # 选课列表

    def enroll(self, course: "Course") -> bool:
        if course in self.courses:
            return False
        self.courses.append(course)
        return True

    def __str__(self):
        course_names = ", ".join(c.name for c in self.courses) or "无"
        return f"[{self.sid}] {self.name} | 课程:{course_names}"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# ===== models/course.py =====
class Course:
    def __init__(self, code: str, name: str, capacity: int):
        self.code = code
        self.name = name
        self.capacity = capacity
        self.students: list["Student"] = []

    def add_student(self, student: "Student") -> bool:
        if len(self.students) >= self.capacity:
            return False
        self.students.append(student)
        return True

    def __str__(self):
        return f"{self.code} {self.name} ({len(self.students)}/{self.capacity})"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# ===== services/enrollment.py =====
from models.student import Student
from models.course import Course

class EnrollmentService:
    @staticmethod
    def enroll(student: Student, course: Course) -> tuple[bool, str]:
        """选课——返回 (成功?, 原因)"""
        if student.enroll(course):     # 检查重复
            if course.add_student(student):  # 检查容量
                return True, f"{student.name} 成功选择 {course.name}"
            else:
                student.courses.remove(course)   # 回滚
                return False, f"{course.name} 已满({course.capacity}人)"
        return False, f"{student.name} 已选过 {course.name}"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# ===== main.py =====
from models.student import Student
from models.course import Course
from services.enrollment import EnrollmentService
from utils.helpers import print_separator

def main():
    print_separator("欢迎使用学生选课系统")

    # 创建课程
    courses = [
        Course("CS101", "Python 编程", 2),
        Course("CS102", "数据结构", 3),
        Course("MATH201", "线性代数", 2),
    ]

    # 创建学生
    students = [
        Student(2024001, "张三"),
        Student(2024002, "李四"),
        Student(2024003, "王五"),
    ]

    # 选课操作
    svc = EnrollmentService()
    operations = [
        (students[0], courses[0]),   # 张三选 Python——成功
        (students[1], courses[0]),   # 李四选 Python——成功
        (students[2], courses[0]),   # 王五选 Python——失败(满)
        (students[0], courses[0]),   # 张三重复选——失败
        (students[0], courses[1]),   # 张三选数据结构——成功
    ]

    for stu, course in operations:
        ok, msg = svc.enroll(stu, course)
        print(f"  {'✅' if ok else '❌'} {msg}")

    print_separator("选课结果")
    for c in courses:
        print(f"  {c}")
    print()
    for s in students:
        print(f"  {s}")
    print_separator("")

if __name__ == "__main__":
    main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
# ===== utils/helpers.py =====
def print_separator(title: str = "", char: str = "=", width: int = 50):
    if title:
        titled = f" {title} "
        pad = (width - len(titled) - len(titled) % 2) // 2
        print(char * pad + titled + char * pad)
    else:
        print(char * width)
1
2
3
4
5
6
7
8

案例知识融合:这个案例展示了一个完整的多模块工程——包结构(models/services/utils)、__init__.py 控制导出接口、相对导入(.student → from .student import Student)、__name__ == '__main__' 入口守卫、前后引用类型注解("Course" 字符串形式)——6 个文件、100+ 行的真实项目骨架。

思考题:

  1. 类型注解中 list["Course"] 为什么写成字符串?直接写 list[Course] 会怎样?(提示:前向引用)
  2. EnrollmentService.enroll 用 @staticmethod——如果改成普通模块级函数,两种设计方案的利弊各是什么?
  3. student.courses.remove(course) 的回滚操作在并发场景下安全吗?为什么 Python 不需要像 Java 那样显式加 synchronized?

# 4.3 文件与异常

# 4.3.1 文件读写操作

# ----- 读文件 -----
# 方式一:一次性读入(小文件)
with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()               # 整个文件→一个字符串

# 方式二:逐行读(大文件——节省内存)
with open("large.log", "r", encoding="utf-8") as f:
    for line in f:                   # 迭代器——每次只读一行
        if "ERROR" in line:
            print(line.strip())

# 方式三:readlines() 读入列表
with open("data.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()            # ['第一行\n', '第二行\n', ...]

# ----- 写文件 -----
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")

# 追加模式
with open("log.txt", "a", encoding="utf-8") as f:
    f.write(f"[{datetime.now()}] 新日志\n")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

文件模式速查表:

模式 含义 文件不存在 文件存在
"r" 读 报错 从头读
"w" 写 创建 清空后写
"a" 追加 创建 末尾追加
"x" 独占创建 创建 报错(防止覆盖)
"r+" 读写 报错 从头
"b" 二进制 — 加在模式后:"rb"、"wb"

# 4.3.2 with 上下文管理器

with 保证资源无论如何都会释放——这是 Python 的 RAII(Resource Acquisition Is Initialization):

# ❌ 没有 with——容易忘记 f.close()
f = open("data.txt", "r")
try:
    content = f.read()
finally:
    f.close()                           # 手动关闭——容易忘

# ✅ 用 with——自动关闭(即使中间抛异常也会关)
with open("data.txt", "r") as f:
    content = f.read()
# f 在这里自动关闭了——不需要写 f.close()
1
2
3
4
5
6
7
8
9
10
11

🔑 with 不是 open() 的专属——任何实现了 __enter__ 和 __exit__ 的对象都能用:

# 自定义上下文管理器
class Timer:
    """统计代码块执行时间"""
    def __enter__(self):
        self.start = time.time()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.time() - self.start
        print(f"⏱ 耗时 {self.elapsed:.4f} 秒")
        return False   # False = 有异常继续传播;True = 吞掉异常

with Timer():
    total = sum(range(10_000_000))   # 代码块结束自动打印耗时
# ⏱ 耗时 0.1234 秒
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 4.3.3 json 与 csv 数据格式

import json

# dict → JSON 字符串(序列化)
data = {
    "name": "张三",
    "age": 25,
    "scores": [85, 92, 78],
    "active": True
}
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(json_str)
# {
#   "name": "张三",
#   "age": 25,
#   "scores": [85, 92, 78],
#   "active": true
# }

# JSON 字符串 → dict(反序列化)
parsed = json.loads(json_str)
print(parsed["name"])          # 张三

# 文件级别操作
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

with open("data.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import csv

# 写 CSV
with open("users.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "age", "city"])
    writer.writeheader()
    writer.writerows([
        {"name": "张三", "age": 25, "city": "深圳"},
        {"name": "李四", "age": 30, "city": "北京"},
    ])

# 读 CSV
with open("users.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['name']},{row['age']}岁,{row['city']}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 4.3.4 try-except-else-finally

Python 的异常机制比 Java/C++ 多了 else 子句:

def safe_divide(a, b):
    try:
        result = a / b
    except ZeroDivisionError:
        print("❌ 不能除以零!")
        return None
    except TypeError as e:
        print(f"❌ 类型错误:{e}")
        return None
    except Exception as e:
        print(f"❌ 未知错误:{type(e).__name__}: {e}")
        return None
    else:
        # try 块没抛任何异常时执行——和 for...else 一样的设计
        print("✅ 计算成功")
        return result
    finally:
        # 无论是否抛异常都执行——收尾工作(关闭文件、释放锁等)
        print("  清理完毕")

print(safe_divide(10, 2))    # ✅ 计算成功 → 清理完毕 → 5.0
print(safe_divide(10, 0))    # ❌ 不能除以零!→ 清理完毕 → None
print(safe_divide(10, "a"))  # ❌ 类型错误 → 清理完毕 → None
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

🔑 except 的匹配顺序:从上往下匹配,命中就停止——所以子类在前、父类在后:

try:
    ...
except ZeroDivisionError:     # 子类——先匹配
    ...
except ArithmeticError:       # 父类——后匹配
    ...
except Exception:             # 兜底——捕获所有
    ...
1
2
3
4
5
6
7
8

# 4.3.5 自定义异常类

class InsufficientFundsError(Exception):
    """余额不足异常"""
    def __init__(self, balance: float, amount: float):
        self.balance = balance
        self.amount = amount
        shortfall = amount - balance
        super().__init__(
            f"余额不足:需要 ¥{amount:,.2f},当前 ¥{balance:,.2f},还差 ¥{shortfall:,.2f}"
        )

class InvalidPinError(Exception):
    """PIN 错误"""
    pass


# 使用
def withdraw(balance: float, amount: float, pin: str):
    if pin != "1234":
        raise InvalidPinError("PIN 码错误,请重试")
    if amount > balance:
        raise InsufficientFundsError(balance, amount)
    return balance - amount

try:
    new_balance = withdraw(100, 200, "1234")
except InsufficientFundsError as e:
    print(f"❌ {e}")
    # ❌ 余额不足:需要 ¥200.00,当前 ¥100.00,还差 ¥100.00
except InvalidPinError as e:
    print(f"❌ {e}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# 4.3.6 综合案例与思考

综合案例:成绩单生成器——文件读写 + JSON + 异常处理

"""
成绩单生成器——文件读写、JSON、CSV、异常处理的综合运用
"""

import json
import csv
import os
from typing import Optional


class StudentRecord:
    def __init__(self, name: str, scores: dict[str, float]):
        self.name = name
        self.scores = scores

    @property
    def total(self) -> float:
        return sum(self.scores.values())

    @property
    def average(self) -> float:
        return self.total / len(self.scores) if self.scores else 0.0

    def to_dict(self) -> dict:
        return {"name": self.name, "scores": self.scores, "total": self.total, "average": round(self.average, 2)}


class ReportGenerator:
    """报告生成器——文件读写 + 异常处理 + 多格式输出"""

    def __init__(self, output_dir: str = "reports"):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def load_records(self, filepath: str) -> list[StudentRecord]:
        """从 JSON 加载成绩记录——带异常处理和格式校验"""
        if not os.path.exists(filepath):
            raise FileNotFoundError(f"文件不存在:{filepath}")

        try:
            with open(filepath, "r", encoding="utf-8") as f:
                data = json.load(f)
        except json.JSONDecodeError as e:
            raise ValueError(f"JSON 格式错误:{e}") from e

        if not isinstance(data, list):
            raise ValueError("JSON 根节点应为数组")

        records = []
        for i, item in enumerate(data, 1):
            try:
                if not isinstance(item, dict):
                    raise ValueError("每项应为对象")
                name = item["name"]
                scores = item["scores"]
                if not isinstance(scores, dict):
                    raise ValueError("scores 应为对象")
                records.append(StudentRecord(name, scores))
            except (KeyError, ValueError) as e:
                print(f"⚠️ 跳过第 {i} 项({e})")
        return records

    def save_json(self, records: list[StudentRecord], filename: str):
        path = os.path.join(self.output_dir, filename)
        data = [r.to_dict() for r in records]
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        print(f"✅ JSON 已保存:{path}")

    def save_csv(self, records: list[StudentRecord], filename: str):
        path = os.path.join(self.output_dir, filename)
        subjects = set()
        for r in records:
            subjects.update(r.scores.keys())
        subjects = sorted(subjects)
        fieldnames = ["name"] + subjects + ["total", "average"]

        with open(path, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            for r in records:
                row = {"name": r.name, **r.scores, "total": r.total, "average": r.average}
                writer.writerow(row)
        print(f"✅ CSV 已保存:{path}")

    def save_report(self, records: list[StudentRecord], filename: str):
        path = os.path.join(self.output_dir, filename)
        records.sort(key=lambda r: r.average, reverse=True)

        with open(path, "w", encoding="utf-8") as f:
            f.write("=" * 55 + "\n")
            f.write(f"{'成绩报告':^55}\n")
            f.write("=" * 55 + "\n")
            header = f"{'排名':<5} {'姓名':<10} {'总分':<8} {'平均':<8}"
            f.write(header + "\n")
            f.write("-" * 55 + "\n")
            for i, r in enumerate(records, 1):
                medal = "🥇" if i == 1 else "🥈" if i == 2 else "🥉" if i == 3 else f"{i:2d}"
                f.write(f"{medal:<5} {r.name:<10} {r.total:<8.1f} {r.average:<8.2f}\n")
            f.write("=" * 55 + "\n")
        print(f"✅ 报告已保存:{path}")


# ===== 演示 =====
def demo():
    # 模拟输入 JSON
    sample_data = [
        {"name": "张三", "scores": {"语文": 85, "数学": 92, "英语": 78}},
        {"name": "李四", "scores": {"语文": 92, "数学": 88, "英语": 95}},
        {"name": "王五", "scores": {"语文": 78, "数学": 85, "英语": 82}},
    ]

    # 写入示例数据
    os.makedirs("data", exist_ok=True)
    with open("data/scores.json", "w", encoding="utf-8") as f:
        json.dump(sample_data, f, ensure_ascii=False, indent=2)

    # 流程
    gen = ReportGenerator()
    try:
        records = gen.load_records("data/scores.json")
        print(f"📥 加载了 {len(records)} 条记录\n")

        gen.save_json(records, "output.json")
        gen.save_csv(records, "output.csv")
        gen.save_report(records, "report.txt")

    except FileNotFoundError as e:
        print(f"❌ {e}")
    except ValueError as e:
        print(f"❌ {e}")
    except Exception as e:
        print(f"❌ 未预期的错误:{type(e).__name__}: {e}")

demo()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

案例知识融合:这个案例覆盖了文件与异常的全部核心——文件读写(open + with)、JSON 序列化/反序列化、CSV DictWriter 输出、os.makedirs/exist_ok 目录创建、try-except 分层捕获(FileNotFoundError → json.JSONDecodeError → ValueError → 兜底 Exception)、raise ... from e 异常链、自定义异常条件——80 行核心逻辑 + 3 种输出格式。

思考题:

  1. raise ValueError(...) from e 中的 from e 是什么作用?如果去掉 from e 会丢失什么信息?
  2. encoding="utf-8-sig" 和 encoding="utf-8" 有什么区别?为什么 CSV 文件推荐用 utf-8-sig?
  3. os.makedirs(output_dir, exist_ok=True) 的参数 exist_ok=True 处理了什么边界条件?去掉它会怎样?

# 4.4 迭代器与生成器

迭代器和生成器是 Python 节省内存的核心工具——处理无限流和大数据集的根基。

# 4.4.1 iter/next 协议

迭代器 = 实现了 __iter__ 和 __next__ 的对象:

# 自定义倒计时迭代器
class Countdown:
    def __init__(self, start: int):
        self.current = start

    def __iter__(self):
        return self                   # 迭代器就是它自己

    def __next__(self):
        if self.current <= 0:
            raise StopIteration       # 迭代结束——for 循环自动捕获
        val = self.current
        self.current -= 1
        return val


# 使用
cd = Countdown(5)
for num in cd:                        # for 循环自动调用 iter() 和 next()
    print(num, end=" ")               # 5 4 3 2 1
print()

# 手动迭代——理解 for 背后的原理
cd2 = Countdown(3)
it = iter(cd2)                        # 等价于 cd2.__iter__()
print(next(it))                       # 3 ← __next__()
print(next(it))                       # 2
print(next(it))                       # 1
# print(next(it))                     # StopIteration!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

🔑 可迭代对象 vs 迭代器:

# 列表是"可迭代对象"(有 __iter__)——但不是"迭代器"(没有 __next__)
lst = [1, 2, 3]
# next(lst)                  # TypeError! 列表不是迭代器

it = iter(lst)               # iter() 返回一个迭代器
print(next(it))              # 1——迭代器有 __next__
1
2
3
4
5
6

# 4.4.2 yield 生成器函数

yield 让函数变成生成器——自动实现 __iter__ 和 __next__,比手动写 Countdown 类简单 10 倍:

# 生成器函数——用 yield 替代 return
def countdown(n: int):
    """倒计时生成器"""
    while n > 0:
        yield n
        n -= 1

# 调用生成器函数——返回生成器对象(不是你要的值!)
g = countdown(5)
print(g)                        # <generator object countdown at 0x...>

print(next(g))                  # 5
print(next(g))                  # 4

for num in countdown(3):        # for 循环消费生成器
    print(num, end=" ")         # 3 2 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

🔑 yield 的执行模型——"暂停-恢复":

def demo_yield():
    print("第 1 步:进入生成器")
    yield 1                     # ← 暂停!返回值 1,等待 next() 唤醒
    print("第 2 步:恢复执行")
    yield 2                     # ← 暂停!返回值 2
    print("第 3 步:再次恢复")
    yield 3
    print("结束")
    # 函数结束 → 自动抛出 StopIteration

g = demo_yield()
print(next(g))
# 第 1 步:进入生成器
# 1
print(next(g))
# 第 2 步:恢复执行
# 2
print(next(g))
# 第 3 步:再次恢复
# 3
# print(next(g))  # 结束 → StopIteration!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

这是 C/C++ 中协程的灵感来源之一——yield 让函数有了"中间状态"。

生成器实战:

# 实战一:逐行读大文件——永远不会 OOM
def read_large_file(filepath: str):
    with open(filepath, "r", encoding="utf-8") as f:
        for line in f:
            yield line.strip()

# 10GB 文件——内存只占一行的大小
for line in read_large_file("huge.log"):
    if "ERROR" in line:
        print(line)

# 实战二:斐波那契无限流
def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

fib = fibonacci()
for _ in range(10):
    print(next(fib), end=" ")   # 0 1 1 2 3 5 8 13 21 34

# 实战三:用 send() 向生成器传值——协程雏形
def accumulator():
    total = 0
    while True:
        value = yield total     # 接收外部传来的值,返回 total
        if value is None:
            break
        total += value

acc = accumulator()
next(acc)                       # 启动——运行到第一个 yield
print(acc.send(10))             # 传入 10 → total=10 → yield 返回 10
print(acc.send(20))             # 传入 20 → total=30 → yield 返回 30
print(acc.send(5))              # 传入 5  → total=35 → yield 返回 35
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# 4.4.3 生成器表达式

# 列表推导式——立即计算,全部装入内存
squares_list = [x ** 2 for x in range(10_000_000)]  # ~80MB

# 生成器表达式——延迟计算,按需产出
squares_gen = (x ** 2 for x in range(10_000_000))   # ~0KB(还没算)

print(next(squares_gen))        # 0(第一个平方——现在才计算)
print(next(squares_gen))        # 1
print(sum(squares_gen))         # 还剩下 999999998 项——但 sum 是逐个消费的!
1
2
3
4
5
6
7
8
9

🔑 括号决定一切:[] → 列表推导式(立即),() → 生成器表达式(延迟):

# sum([]) = 全部算完再求和——内存炸
print(sum([x*2 for x in range(10_000_000)]))   # 先生成 80MB 列表,再求和

# sum(()) = 一边生成一边求和——内存恒定为常量
print(sum(x*2 for x in range(10_000_000)))     # 生成器——再大的 range 也不怕
1
2
3
4
5

# 4.4.4 itertools 常用工具

from itertools import *

# count()——无限计数
for i, val in zip(count(start=100, step=5), "ABC"):
    print(f"{i}:{val}", end="  ")     # 100:A  105:B  110:C

# cycle()——循环往复
for i, val in zip(range(6), cycle("RGB")):
    print(val, end=" ")               # R G B R G B

# chain()——串联多个迭代器
print(list(chain([1,2], "ab", (10,20))))  # [1, 2, 'a', 'b', 10, 20]

# combinations() / permutations()
from itertools import combinations, permutations
print(list(combinations("ABC", 2)))      # [('A','B'), ('A','C'), ('B','C')]
print(list(permutations("ABC", 2)))      # [('A','B'), ('A','C'), ('B','A'), ...]

# groupby()——按键分组(需预排序)
data = [("A", 1), ("A", 2), ("B", 3), ("B", 4)]
for key, group in groupby(data, key=lambda x: x[0]):
    items = list(group)
    print(f"{key}: {items}")             # A: [('A',1), ('A',2)]  B: [('B',3), ('B',4)]

# islice()——切片但不创建列表
print(list(islice(range(100), 10, 20, 2)))  # [10, 12, 14, 16, 18]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# 4.4.5 生成器节省内存原理

import sys

# 列表:全部在内存中
nums_list = [x for x in range(1_000_000)]
print(f"列表内存:{sys.getsizeof(nums_list) / 1024 / 1024:.1f} MB")

# 生成器:仅一个对象在内存中
nums_gen = (x for x in range(1_000_000))
print(f"生成器内存:{sys.getsizeof(nums_gen):,} 字节")   # ~200 字节!

# range 甚至更省——它不存储元素,只记住 start/stop/step
r = range(1_000_000_000)    # 10 亿——毫秒级创建
print(f"range 内存:{sys.getsizeof(r):,} 字节")           # 48 字节!
1
2
3
4
5
6
7
8
9
10
11
12
13

🔑 什么时候用生成器 vs 列表:

场景 选择 原因
数据只遍历一次 生成器 不用一次装进内存
需要索引访问 lst[i] 列表 生成器不支持随机访问
需要多次遍历 列表(或转 list(gen)) 生成器消费一次就耗尽
无限序列 生成器 列表不可能装下无限
需要 len() 列表 生成器没有长度

# 4.4.6 综合案例与思考

综合案例:日志流处理器——生成器的链式管道

"""
日志流处理器——生成器链式管道,演示"延迟求值"的力量

处理真实场景:10GB Nginx 日志 → 过滤 → 提取 → 聚合
"""

from itertools import islice, groupby
from collections import Counter
import re
from datetime import datetime


# ① 模拟无限日志流(实际项目中可以是文件尾随 / kafka 流)
def log_stream():
    """生成器:模拟 Nginx 日志流——一行一行产出"""
    sample_lines = [
        '192.168.1.1 - - [08/Jun/2025:14:00:01 +0800] "GET /api/users HTTP/1.1" 200 1234 0.012',
        '10.0.0.5 - - [08/Jun/2025:14:00:02 +0800] "POST /api/login HTTP/1.1" 401 88 0.089',
        '192.168.1.2 - - [08/Jun/2025:14:00:03 +0800] "GET /api/products HTTP/1.1" 200 5678 0.034',
        '172.16.0.1 - - [08/Jun/2025:14:00:04 +0800] "GET /static/style.css HTTP/1.1" 304 0 0.001',
        '192.168.1.1 - - [08/Jun/2025:14:00:05 +0800] "PUT /api/users/3 HTTP/1.1" 500 234 1.234',
        '10.0.0.5 - - [08/Jun/2025:14:00:06 +0800] "GET /api/users HTTP/1.1" 200 1234 0.008',
        '192.168.1.3 - - [08/Jun/2025:14:00:07 +0800] "GET /api/search?q=python HTTP/1.1" 200 9234 0.245',
    ]
    for line in sample_lines:
        yield line.strip()


# ② 解析阶段:原始文本 → 结构化 dict
def parse_log(lines):
    """生成器:每行日志 → 解析后的 dict"""
    pattern = re.compile(
        r'(?P<ip>\S+).*?\[(?P<time>[^\]]+)\]\s+"(?P<method>\S+) (?P<path>\S+).*?" '
        r'(?P<status>\d{3}) (?P<bytes>\d+)(?: (?P<rt>[\d.]+))?'
    )
    for line in lines:
        m = pattern.match(line)
        if m:
            yield {
                "ip": m["ip"],
                "time": m["time"],
                "method": m["method"],
                "path": m["path"],
                "status": int(m["status"]),
                "bytes": int(m["bytes"]),
                "response_time": float(m.get("rt") or 0),
            }


# ③ 过滤阶段:只保留关注的数据
def filter_errors(entries, min_status: int = 400):
    """生成器:过滤错误请求"""
    for entry in entries:
        if entry["status"] >= min_status:
            yield entry


def filter_api(entries):
    """生成器:只保留 API 请求"""
    for entry in entries:
        if entry["path"].startswith("/api/"):
            yield entry


# ④ 处理:限制 + 聚合
def analyze(entries):
    """消费生成器——计算统计信息"""
    entries = list(entries)  # 生成器已过滤到很小——安全地转为列表
    if not entries:
        return

    # 状态码分布
    status_count = Counter(e["status"] for e in entries)
    # 平均响应时间
    avg_rt = sum(e["response_time"] for e in entries) / len(entries)
    # 总流量
    total_bytes = sum(e["bytes"] for e in entries)
    # 慢请求 TOP 3
    slow = sorted(entries, key=lambda e: e["response_time"], reverse=True)[:3]

    print(f"\n{'='*55}")
    print(f"{'日志分析结果':^55}")
    print(f"{'='*55}")
    print(f"\n📊 状态码分布:")
    for code, cnt in sorted(status_count.items()):
        name = {200: "OK", 304: "Not Modified", 401: "Unauthorized", 500: "Server Error"}.get(code, "?")
        print(f"  {code} {name:<14} {cnt} 次")

    print(f"\n⏱ 平均响应时间:{avg_rt:.4f}s")
    print(f"📦 总流量:{total_bytes:,} 字节")

    print(f"\n🐌 慢请求 TOP 3:")
    for i, e in enumerate(slow, 1):
        print(f"  {i}. {e['response_time']:.3f}s  [{e['status']}] {e['path']}")
    print(f"{'='*55}\n")


# ===== 管道运行 =====
print("=" * 55)
print(f"{'日志流处理管道':^55}")
print("=" * 55)
print("日志源 → 解析 → 过滤(API) → 过滤(错误) → 分析\n")

# 链式调用——每一环都是生成器,数据只在最终消费时才流动
pipeline = filter_api(parse_log(log_stream()))  # 先过滤出 API
errors = filter_errors(pipeline, min_status=400)  # 再过滤错误
analyze(errors)                                     # 消费

# 再看正常请求的统计
print("(正常 API 请求统计)")
all_api = filter_api(parse_log(log_stream()))
normal = filter(
    lambda e: e["status"] < 400,
    all_api
)
analyze(list(normal))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

案例知识融合:这个案例展示了 Python 生成器的精髓——管道模式。log_stream→parse_log→filter_api→filter_errors→analyze 五环组成一条流水线,每环都是生成器——数据只在 analyze 最终 list() 时才实际流入。对于 10GB 日志,内存占用恒定为"一行的大小",比"先读全文件再过滤"省了 10000+ 倍内存。

思考题:

  1. 生成器管道中,每一环的数据产出时机是什么?如果 analyze 永远不调用 next(),前面的 log_stream 会执行哪怕一行吗?
  2. filter(all_api, lambda e: e["status"] < 400) 返回的是迭代器——但如果传入的 all_api 已经是列表(不是生成器),filter 的行为有区别吗?
  3. itertools.groupby 要求数据预排序——为什么?如果传给未排序的数据会发生什么?

# 4.5 新手陷阱

# 陷阱 说明
1 类属性被实例属性遮蔽 self.attr = val 只在实例上创建——不会改类属性,但会让 self.attr 看不到类属性
2 可变类属性共享 class A: items = [] ——所有实例共享同一个列表,改一个全变。应放 __init__ 里
3 没有 with 打开文件 f = open(...) 忘记 f.close() ——用 with 自动关闭
4 生成器只能消费一次 g = (x for x in range(5)); list(g); list(g) → 第二次返回 []——耗尽后不可复用
5 except 捕获太过宽泛 except Exception 吞掉了所有错误——至少打印日志,不要空 except:

陷阱 4 详解:

g = (x * 2 for x in range(5))
print(list(g))             # [0, 2, 4, 6, 8]
print(list(g))             # []——生成器已经耗尽了!

# ✅ 如果需要多次遍历——转为列表
items = list(g)            # 或:用函数重新创建生成器
1
2
3
4
5
6

# 4.6 综合思考题

  1. Python 的 class 也是对象:Student 这个类本身也是 type 的实例——isinstance(Student, type) 返回 True。这意味着你可以在运行时动态创建类(type('NewClass', (Base,), {'attr': 1}))。这种"元编程"能力在什么场景下有用——Django 的 ORM 是如何利用元类实现 class User(models.Model) → 自动生成数据库建表语句的?

  2. __slots__:默认情况下,每个 Python 对象的 __dict__ 字典存储了所有属性——灵活但有内存开销。class Point: __slots__ = ('x', 'y') 可以禁用 __dict__,让内存占用从 ~300 字节降到 ~50 字节——但代价是不能再动态添加属性。什么时候值得用 __slots__?如果你有 100 万个 Point 对象,它能省多少内存?

  3. 生成器 vs 协程:yield 的 send() 方法让生成器成了单向通道的协程。Python 3.5 引入 async/await 后,yield from 被 await 取代。yield 模拟的协程和原生 async/await 有什么区别?为什么后者是未来?

  4. 文件编码的血泪史:open(..., encoding="utf-8") 在 Windows 上不写 encoding 参数会默认用 GBK——导致中文字符乱码。Python 3 已经全面使用 Unicode,为什么文件 I/O 还得显式指定编码?如果 Python 默认全部用 UTF-8,会有什么兼容性问题?

  5. 异常链与调试:raise NewError(...) from original_error 会保留完整的异常链——__cause__ 和 __context__。在复杂的多层系统中,异常链信息如何帮助快速定位根因?except ...: raise 和 except ...: raise NewError 在 traceback 上有何区别?


# 4.7 Python 基础入门知识地图

四章学完,你应当能在一张图上看到所有知识点的位置和连接关系:

第 1 章  入门与基础类型                   第 2 章  序列与集合类型
┌──────────────────────┐          ┌────────────────────────────┐
│ Python 安装 + REPL   │          │  字符串进阶(切片/格式化)   │
│ 变量(引用 vs 值)    │──────→──│  列表(推导式/排序/深浅拷贝)│
│ 数字/字符串/布尔/None │          │  元组(不可变/Packing)     │
│ 类型检查与转换        │          │  字典(哈希/遍历/get)      │
│ 运算符(算术/比较/逻辑)│         │  集合(去重/交并差/hash)  │
│ 输入输出 print/input  │          │  正则初探                   │
└──────────────────────┘          └────────────────────────────┘
         │                                      │
         └──────────── 门禁 ─────────────┘
                    │
                    ▼
第 3 章  流程控制与函数                   第 4 章  面向对象与工程
┌──────────────────────┐          ┌──────────────────────────────┐
│ if/elif/else 分支     │          │  class/对象(self/__init__)  │
│ for 迭代 + range      │          │  继承与多态(MRO)            │
│ while 条件循环        │          │  封装(_ / __ / @property)   │
│ break/continue/pass   │          │  魔术方法(__str__/__len__…) │
│ for...else 独特语法   │          │  模块/包/虚拟环境/pip         │
│ 海象运算符 :=         │──────→──│  文件读写 + with 上下文       │
│ def 函数定义/多值返回  │          │  JSON / CSV 数据处理          │
│ 五种参数类型全景      │          │  try-except-else-finally     │
│ LEGB 作用域规则       │          │  自定义异常                   │
│ global / nonlocal     │          │  迭代器(__iter__/__next__)  │
│ 闭包 = 函数 + 捕获变量 │          │  生成器(yield/表达式/内存)  │
│ lambda 匿名函数       │          │  itertools 工具箱             │
│ @装饰器 (NEW ✨)      │          │  生成器管道模式               │
└──────────────────────┘          └──────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

学习路线建议:

阶段 内容 时间
①入门 第1章 §1.1~§1.4 4h
②地基 第2章 §2.1~§2.3 5h
③控制 第3章 §3.1~§3.5 6h
④进阶 第3章 §3.6 + 第4章 §4.1~§4.4 8h

关键连接线:

  • §1.2.7 引用 vs 值 → §2.2.3 深浅拷贝 → §3.3.2 默认参数陷阱:同一条"可变对象共享"的线贯穿三章
  • §3.4 闭包 → §3.5 lambda → §3.6 装饰器:闭包是装饰器的"发动机"
  • §3.6 装饰器 → §4.1 @property/@staticmethod/@classmethod:装饰器语法在 OOP 中的集中体现
  • §4.4 生成器 → §4.4.6 管道模式:函数式编程的延迟求值思想

🎓 四章学完后你能做什么:

  • 独立完成 500 行以下的 Python 脚本(数据分析、文件处理、命令行工具)
  • 读懂开源 Python 项目的类定义、装饰器、生成器用法
  • 具备进入 Flask/Django、NumPy/Pandas、pytest 等框架的基础能力
  • 最重要的是——你能说出"这段 Python 代码底层在做什么"——这远超"会用"的层面
#Python#基础
上次更新: 2026/06/17, 12:47:39
流程控制与函数
爬虫全流程实战

← 流程控制与函数 爬虫全流程实战→

最近更新
01
信号崩溃快速排查
06-15
02
CoreDump破案
06-15
03
perf火焰图实战
06-15
更多文章>
Theme by Vdoing | Copyright © 2019-2026 杨充 | MIT License | 桂ICP备2024034950号 | 桂公网安备45142202000030
  • 跟随系统
  • 浅色模式
  • 深色模式
  • 阅读模式