编程进阶网 编程进阶网
首页
  • 在线工具
  • JSON工具
  • 文本工具
  • 图片处理
  • 文档转化
  • 代码压缩
  • 加解密
  • 时间日期
  • 网络工具
  • 颜色设计
  • 二维码
  • 开发实用
  • 计算机组成原理
  • 操作系统原理
  • 网络协议原理
  • 数据库系统原理
  • 序卷导读
  • 数据本质
  • 运行模型
  • 并发设计
  • 内存真相
  • 交互系统
  • 面向对象
  • 设计原则
  • 设计模式
  • 系统架构
  • 体系建设
  • 代码品质
  • 方案设计
  • 稳定可靠
  • 工程运维
  • 性能优化
  • 数据结构导论
  • 线性结构详解
  • 树哈希结构论
  • 容器设计实战
  • 经典算法思想
  • 工程案例剖析
  • 算法题库精练
  • C语言入门
  • C综合案例
  • C专栏博客
  • C标准集库
  • C++入门教程
  • C++综合案例
  • C++专栏博客
  • C++编程技巧
  • Java入门教程
  • Java综合案例
  • Java专栏博客
  • Go入门教程
  • Go综合案例
  • Go专栏博客
  • Go开发技巧
  • JavaScript入门
  • JavaScript案例
  • JavaScript高级
  • Android库解读
  • Android专栏
  • iOS ObjC入门
  • iOS Swift入门
  • iOS入门精通
  • Web之Html手册
  • Web之TypeScript
  • Web之Vue高级进阶
  • Linux之QML入门
  • Linux之QT核心库
  • Python教程
  • Shell&Bash教程
  • 工具脚本
  • 自动化脚本
  • 质量保障
  • 产品思考
  • 软实力
  • 开发流程
  • Git应用
  • 技术模版
  • 技术规范
  • Markdown
  • Mermaid
  • 开源协议
  • 毛选解读
  • 自我精进
  • 关于我
  • 自我精进
  • 职场管理
  • 职场面试
  • 心情杂货
  • 友情链接

杨充

专注编程 · 终身学习者
首页
  • 在线工具
  • JSON工具
  • 文本工具
  • 图片处理
  • 文档转化
  • 代码压缩
  • 加解密
  • 时间日期
  • 网络工具
  • 颜色设计
  • 二维码
  • 开发实用
  • 计算机组成原理
  • 操作系统原理
  • 网络协议原理
  • 数据库系统原理
  • 序卷导读
  • 数据本质
  • 运行模型
  • 并发设计
  • 内存真相
  • 交互系统
  • 面向对象
  • 设计原则
  • 设计模式
  • 系统架构
  • 体系建设
  • 代码品质
  • 方案设计
  • 稳定可靠
  • 工程运维
  • 性能优化
  • 数据结构导论
  • 线性结构详解
  • 树哈希结构论
  • 容器设计实战
  • 经典算法思想
  • 工程案例剖析
  • 算法题库精练
  • C语言入门
  • C综合案例
  • C专栏博客
  • C标准集库
  • C++入门教程
  • C++综合案例
  • C++专栏博客
  • C++编程技巧
  • Java入门教程
  • Java综合案例
  • Java专栏博客
  • Go入门教程
  • Go综合案例
  • Go专栏博客
  • Go开发技巧
  • JavaScript入门
  • JavaScript案例
  • JavaScript高级
  • Android库解读
  • Android专栏
  • iOS ObjC入门
  • iOS Swift入门
  • iOS入门精通
  • Web之Html手册
  • Web之TypeScript
  • Web之Vue高级进阶
  • Linux之QML入门
  • Linux之QT核心库
  • Python教程
  • Shell&Bash教程
  • 工具脚本
  • 自动化脚本
  • 质量保障
  • 产品思考
  • 软实力
  • 开发流程
  • Git应用
  • 技术模版
  • 技术规范
  • Markdown
  • Mermaid
  • 开源协议
  • 毛选解读
  • 自我精进
  • 关于我
  • 自我精进
  • 职场管理
  • 职场面试
  • 心情杂货
  • 友情链接
  • 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)          # 北京大学

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

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

# 等价写法:类名.方法(实例)
s1 = Student("张三", 20)
print(Student.introduce(s1))  # 和 s1.introduce() 完全等价

# 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())              # 小黄:汪汪! 我喜欢玩骨头!

🔑 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

⚠️ 多重继承很危险——菱形继承会导致"某个父类的 __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 哲学:我们都是成年人)

🔑 名称改写规则:__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

🔑 对比 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);   ← 方法调用的语法,暴露了"是方法"
# Python:@property 封装后,语法和属性一样
t.celsius          # 读——看不出是方法
t.celsius = 100    # 写——看不出是方法
# 这就是 Python 的"统一访问原则"——接口稳定不变,内部可以随时改成属性或方法

# 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²)

常用特殊方法速查表:

方法 触发条件 类比
__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()

运行片段:

🏦 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 笔

案例知识融合:这个案例在真实场景中综合运用了 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 *         # ❌ 大量符号涌入,看不出来源

🔑 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
# main.py——导入 my_utils
import my_utils

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

🔑 原理:当用 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                  # 打包配置

__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())

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

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

# 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

⚠️ 永远不要 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
# ===== 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}"
# ===== 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})"
# ===== 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}"
# ===== 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()
# ===== 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)

案例知识融合:这个案例展示了一个完整的多模块工程——包结构(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")

文件模式速查表:

模式 含义 文件不存在 文件存在
"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()

🔑 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 秒

# 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)
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']}")

# 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

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

try:
    ...
except ZeroDivisionError:     # 子类——先匹配
    ...
except ArithmeticError:       # 父类——后匹配
    ...
except Exception:             # 兜底——捕获所有
    ...

# 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}")

# 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()

案例知识融合:这个案例覆盖了文件与异常的全部核心——文件读写(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!

🔑 可迭代对象 vs 迭代器:

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

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

# 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

🔑 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!

这是 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

# 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 是逐个消费的!

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

# 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 也不怕

# 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]

# 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 字节!

🔑 什么时候用生成器 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))

案例知识融合:这个案例展示了 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)            # 或:用函数重新创建生成器

# 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章 §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/28, 17:55:19
流程控制与函数
爬虫全流程实战

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

最近更新
01
科学方法实践论法
06-28
02
辩证思维矛盾论法
06-28
03
毛选中的调查观念
06-28
更多文章>
Theme by Vdoing | Copyright © 2019-2026 杨充 | MIT License | 鄂ICP备2024073355号-1 | 鄂ICP备2024073355号
  • 跟随系统
  • 浅色模式
  • 深色模式
  • 阅读模式