Python教程 — 装饰器、生成器与文件IO

> 上节课搞定了面向对象，这节课我们深入 Python 的几个杀手级特性：装饰器、生成器、文件操作和异常处理。这四个概念掌握了，你的 Python 水平就能从"入门"升级到"能用"了。

---

## 一、装饰器（Decorator）—— 函数的"外衣"

### 1.1 基本概念

```python

import time

# 不带参数的装饰器：计算函数运行时间

deftimer(func):

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

defwrapper(*args, **kwargs):

        start = time.time()

        result = func(*args, **kwargs)

        end = time.time()

print(f"{func.__name__} 执行耗时: {end - start:.4f}秒")

return result

return wrapper

@timer

defslow_function():

"""模拟一个慢函数"""

    time.sleep(1)

return"完成！"

slow_function()

# 输出：slow_function 执行耗时: 1.0012秒

```

### 1.2 带参数的装饰器

```python

import functools

defrepeat(n_times):

"""带参数的装饰器：重复执行函数"""

defdecorator(func):

@functools.wraps(func)  # 保留原函数的元信息

defwrapper(*args, **kwargs):

for _ inrange(n_times):

                result = func(*args, **kwargs)

return result

return wrapper

return decorator

@repeat(n_times=3)

defgreet(name):

print(f"Hello, {name}!")

greet("Alice")

# 输出三次：Hello, Alice!

```

### 1.3 装饰器的常见应用场景

```python

import functools

# ====== 1. 缓存装饰器（记忆化）======

defmemoize(func):

"""缓存函数结果，避免重复计算"""

    cache = {}

@functools.wraps(func)

defwrapper(*args):

if args notin cache:

            cache[args] = func(*args)

return cache[args]

return wrapper

@memoize

deffibonacci(n):

if n < 2:

return n

return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(50))  # 非常快！（缓存了中间结果）

# ====== 2. 权限校验装饰器 ======

defrequire_role(role):

"""检查用户角色"""

defdecorator(func):

@functools.wraps(func)

defwrapper(user, *args, **kwargs):

if user.get("role") == role or user.get("role") == "admin":

return func(user, *args, **kwargs)

else:

raisePermissionError(f"需要 {role} 权限！")

return wrapper

return decorator

# 模拟用户

admin_user = {"name": "张三", "role": "admin"}

guest_user = {"name": "李四", "role": "guest"}

@require_role("admin")

defdelete_database(user):

print(f"✅ {user['name']} 执行了删除数据库操作")

delete_database(admin_user)       # ✅ 通过

# delete_database(guest_user)    # ❌ PermissionError

```

### 1.4 多个装饰器叠加

```python

deflogging_decorator(func):

@functools.wraps(func)

defwrapper(*args, **kwargs):

print(f"[LOG] 调用 {func.__name__}")

        result = func(*args, **kwargs)

print(f"[LOG] {func.__name__} 返回: {result}")

return result

return wrapper

defvalidate_input(func):

@functools.wraps(func)

defwrapper(*args, **kwargs):

for arg in args:

ifisinstance(arg, str) andnot arg:

raiseValueError("不允许空字符串参数")

return func(*args, **kwargs)

return wrapper

@logging_decorator

@validate_input

defdivide(a, b):

return a / b

print(divide(10, 2))  

# [LOG] 调用 divide

# [LOG] divide 返回: 5.0

```

---

## 二、生成器（Generator）—— 省内存的数据流

### 2.1 yield 关键字

```python

# 普通函数 vs 生成器函数

# 普通函数：一次性返回整个列表

defget_evens_normal(n):

    evens = []

for i inrange(n):

if i % 2 == 0:

            evens.append(i)

return evens  # 全部放在内存里

# 生成器函数：逐个产出数据

defget_evens_gen(n):

for i inrange(n):

if i % 2 == 0:

yield i  # 每次产出一个值，暂停在这里

# 使用

print(list(get_evens_normal(10)))   # [0, 2, 4, 6, 8]

print(list(get_evens_gen(10)))       # [0, 2, 4, 6, 8]

```

>**关键区别**：

>- 普通函数用 `return`，一次返回所有结果

>- 生成器用 `yield`，惰性求值，需要时才计算

>- 对于大数���据集（百万级），生成器节省大量内存

### 2.2 生成器表达式

```python

# 类似列表推导式，但是用圆括号

squares = (x**2for x inrange(1000000))  # 几乎不占内存！

# 而不是

squares = [x**2for x inrange(1000000)]   # 占用几百MB

# 逐个消费生成器

print(next(squares))  # 0

print(next(squares))  # 1

print(next(squares))  # 4

# 配合 sum() 等函数使用

total = sum(x**2for x inrange(1000000))  # 高效！

```

### 2.3 实用的生成器示例

```python

import itertools

# 无限生成器

defcountdown(n):

"""倒计时"""

while n > 0:

yield n

        n -= 1

defnaturals():

"""自然数序列（无限）"""

    n = 1

whileTrue:

yield n

        n += 1

# 从 naturals 中取前 10 个

for i in itertools.islice(naturals(), 10):

print(i, end=" ")  # 1 2 3 4 5 6 7 8 9 10

# 数据流水线：像管道一样加工数据

defread_lines(filename):

"""逐行读取文件"""

withopen(filename, "r", encoding="utf-8") as f:

for line in f:

yield line.strip()

deffilter_non_empty(lines):

"""过滤空行"""

for line in lines:

if line:

yield line

defnormalize(lines):

"""统一转换为小写"""

for line in lines:

yield line.lower()

# 串联使用

# lines = normalize(filter_non_empty(read_lines("data.txt")))

# for line in lines:

#     print(line)

```

### 2.4 生成器发送值

```python

defaccumulator():

"""累加器：接收值并产出累计和"""

    total = 0

whileTrue:

        value = yield total

if value isNone:

break

        total += value

acc = accumulator()

next(acc)  # 启动生成器（预读）

print(acc.send(10))  # 10

print(acc.send(20))  # 30

print(acc.send(15))  # 45

acc.close()

```

---

## 三、文件操作（File IO）

### 3.1 文本文件读写

```python

# ====== 写入文件 ======

text = """

Python 是一门优秀的编程语言。

它的语法简洁，适合初学者入门。

2026 年 Python 依然是最热门的语言之一。

"""

withopen("demo.txt", "w", encoding="utf-8") as f:

    f.write(text)

print("✅ 文件已写入")

# ====== 读取文件 ======

withopen("demo.txt", "r", encoding="utf-8") as f:

    content = f.read()

print(content)

# ====== 逐行读取 ======

withopen("demo.txt", "r", encoding="utf-8") as f:

for line in f:

print(line.strip())

# ====== 读取所有行到列表 ======

withopen("demo.txt", "r", encoding="utf-8") as f:

    lines = f.readlines()

print(f"共 {len(lines)} 行")

```

>**关键**：始终使用 `with` 语句！它会自动关闭文件，即使发生异常。

### 3.2 文件操作模式

| 模式 | 说明 |

|------|------|

| `r` | 只读（默认），文件不存在时报错 |

| `w` | 只写，文件不存在则创建，存在则清空 |

| `a` | 追加，文件不存在则创建 |

| `r+` | 读写 |

| `rb` / `wb` | 二进制模式（用于图片等） |

### 3.3 CSV 文件处理

```python

import csv

# 写入 CSV

data = [

    ["姓名", "年龄", "城市"],

    ["张三", 25, "北京"],

    ["李四", 30, "上海"],

    ["王五", 28, "广州"],

]

withopen("people.csv", "w", encoding="utf-8", newline="") as f:

    writer = csv.writer(f)

    writer.writerows(data)

# 读取 CSV

withopen("people.csv", "r", encoding="utf-8") as f:

    reader = csv.reader(f)

    header = next(reader)  # 跳过表头

for row in reader:

print(f"{row[0]}, {row[1]}岁, 来自{row[2]}")

# 使用 DictReader（更友好）

withopen("people.csv", "r", encoding="utf-8") as f:

    reader = csv.DictReader(f)

for row in reader:

print(f"{row['姓名']} - {row['年龄']}岁")

```

### 3.4 JSON 文件处理

```python

import json

# 写入 JSON

data = {

"name": "张三",

"age": 25,

"hobbies": ["阅读", "编程", "跑步"],

"address": {"city": "北京", "zip": "100000"}

}

withopen("data.json", "w", encoding="utf-8") as f:

    json.dump(data, f, ensure_ascii=False, indent=2)

# ensure_ascii=False: 中文正常显示

# indent=2: 格式化缩进

# 读取 JSON

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

    loaded = json.load(f)

print(loaded["name"])        # 张三

print(loaded["address"]["city"])  # 北京

```

---

## 四、异常处理（Exception Handling）

### 4.1 基本结构

```python

try:

# 可能出错的代码

    numerator = int(input("请输入分子: "))

    denominator = int(input("请输入分母: "))

    result = numerator / denominator

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

exceptValueError:

print("❌ 请输入有效的整数！")

exceptZeroDivisionError:

print("❌ 分母不能为零！")

exceptExceptionas e:

print(f"❌ 未知错误: {e}")

else:

# 没有异常时执行

print("✅ 计算成功！")

finally:

# 无论是否异常都会执行（清理工作）

print("🔄 程序结束")

```

### 4.2 自定义异常

```python

classAgeError(ValueError):

"""年龄错误异常"""

pass

classBankAccount:

def__init__(self, owner, balance):

if balance < 0:

raiseValueError("余额不能为负数")

self.owner = owner

self.balance = balance

defwithdraw(self, amount):

if amount <= 0:

raiseValueError("取款金额必须为正数")

if amount > self.balance:

raise InsufficientFundsError(f"余额不足！需要 ¥{amount}，当前 ¥{self.balance}")

self.balance -= amount

return amount

classInsufficientFundsError(Exception):

"""余额不足异常"""

pass

# 使用

try:

    account = BankAccount("张三", 1000)

    account.withdraw(1500)

except InsufficientFundsError as e:

print(f"💸 {e}")

exceptValueErroras e:

print(f"参数错误: {e}")

```

### 4.3 断言（assert）

```python

# 用于开发调试，不符合条件就抛出 AssertionError

defcalculate_average(numbers):

assertlen(numbers) > 0, "列表不能为空"

assertall(isinstance(n, (int, float)) for n in numbers), "所有元素必须是数字"

returnsum(numbers) / len(numbers)

print(calculate_average([85, 90, 92]))  # 89.0

# calculate_average([])  # AssertionError: 列表不能为空

```

---

## 五、综合实战：日志文件分析器

把装饰器、生成器、文件IO、异常处理串在一起：

```python

import json

from datetime import datetime

deftime_it(func):

"""计时装饰器"""

import time

defwrapper(*args, **kwargs):

        start = time.time()

        result = func(*args, **kwargs)

        elapsed = time.time() - start

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

return result

return wrapper

defparse_log_line(line):

"""解析日志行（生成器）"""

    line = line.strip()

ifnot line:

return

try:

        parts = line.split("|")

iflen(parts) != 4:

return

        timestamp = parts[0].strip()

        level = parts[1].strip()

        module = parts[2].strip()

        message = parts[3].strip()

if level notin ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):

return

yield {

"timestamp": timestamp,

"level": level,

"module": module,

"message": message

        }

exceptException:

return

defgenerate_logs(filename):

"""逐行读取并解析日志（生成器管道）"""

withopen(filename, "r", encoding="utf-8") as f:

for line in f:

            parsed = parse_log_line(line)

if parsed:

yield parsed

@time_it

defanalyze_logs(filename):

"""分析日志文件"""

    logs = list(generate_logs(filename))

# 统计

    levels = {}

    modules = {}

    errors = []

for log in logs:

        level = log["level"]

        module = log["module"]

        levels[level] = levels.get(level, 0) + 1

        modules[module] = modules.get(module, 0) + 1

if level in ("ERROR", "CRITICAL"):

            errors.append(log)

# 输出报告

    report = {

"total": len(logs),

"levels": levels,

"modules": modules,

"error_count": len(errors),

"errors": [

            {"timestamp": e["timestamp"], "message": e["message"]}

for e in errors[:10]  # 最多显示前10条错误

        ]

    }

# 打印摘要

print(f"\n{'='*50}")

print(f"  日志分析报告")

print(f"{'='*50}")

print(f"  总记录数: {report['total']}")

print(f"\n  级别分布:")

for level, count insorted(report['levels'].items()):

        bar = "█" * count // 2

print(f"    {level:8s}{count:4d}{bar}")

print(f"\n  模块分布:")

for module, count insorted(report['modules'].items(), key=lambdax: x[1], reverse=True):

print(f"    {module:15s}{count:4d}")

if errors:

print(f"\n  ⚠️ 错误日志（共 {len(errors)} 条）:")

for err in report["errors"]:

print(f"    [{err['timestamp']}] {err['message']}")

print(f"{'='*50}\n")

# 保存报告

withopen("report.json", "w", encoding="utf-8") as f:

        json.dump(report, f, ensure_ascii=False, indent=2)

print("📄 报告已保存到 report.json")

return report

# 创建一个示例日志文件供测试

sample_logs = """

2026-07-02 10:00:01|INFO|app|系统启动完成

2026-07-02 10:00:02|DEBUG|db|连接数据库成功

2026-07-02 10:01:00|INFO|auth|用户登录: admin

2026-07-02 10:02:30|WARNING|cache|缓存命中率下降到 65%

2026-07-02 10:03:00|ERROR|api|请求超时: /api/users

2026-07-02 10:05:00|INFO|app|定时任务执行完成

2026-07-02 10:10:00|ERROR|db|数据库连接断开

2026-07-02 10:10:01|CRITICAL|app|服务降级，切换到备用数据库

2026-07-02 10:15:00|INFO|app|恢复正常运行

2026-07-02 10:20:00|WARNING|api|API 调用频率接近上限

"""

withopen("app.log", "w", encoding="utf-8") as f:

    f.write(sample_logs.strip())

analyze_logs("app.log")

```

运行效果：

```

[analyze_logs] 耗时: 0.002秒

==================================================

  日志分析报告

==================================================

  总记录数: 10

  级别分布:

    DEBUG       1  █

    INFO        5  █████

    WARNING     2  ██

    ERROR       2  ██

    CRITICAL    1  █

  模块分布:

    app             4

    api             2

    db              2

    auth            1

    cache           1

  ⚠️ 错误日志（共 3 条）:

    [2026-07-02 10:03:00] 请求超时: /api/users

    [2026-07-02 10:10:00] 数据库连接断开

    [2026-07-02 10:10:01] 服务降级，切换到备用数据库

==================================================

📄 报告已保存到 report.json

```

---

## 六、本节要点总结

| 概念 | 核心要点 |

|------|---------|

| **装饰器** | `@decorator` 语法糖，不修改原函数扩展功能，`@functools.wraps` 保元数据 |

| **生成器** | `yield` 惰性求值，省内存，生成器表达式 `(x**2 for x in range(10))` |

| **文件IO** | 用 `with` 语句自动管理文件、CSV 用 `csv` 模块、JSON 用 `json` 模块 |

| **异常处理** | `try-except-else-finally` 完整结构，自定义异常类 |

---

## 七、练习题

### 练习 1：用装饰器实现重试机制

```python

import functools

import time

defretry(max_retries=3, delay=1):

"""失败重试装饰器"""

defdecorator(func):

@functools.wraps(func)

defwrapper(*args, **kwargs):

for attempt inrange(1, max_retries + 1):

try:

return func(*args, **kwargs)

exceptExceptionas e:

print(f"第 {attempt} 次尝试失败: {e}")

if attempt < max_retries:

                        time.sleep(delay)

else:

raise

return wrapper

return decorator

@retry(max_retries=3, delay=0.5)

defunstable_api():

import random

if random.random() < 0.7:

raiseConnectionError("网络超时")

return"请求成功！"

print(unstable_api())

```

### 练习 2：用生成器实现数据过滤器

```python

defpositive_numbers(nums):

"""只产出正数"""

for n in nums:

if n > 0:

yield n

defsquared(nums):

"""只产出平方"""

for n in nums:

yield n ** 2

deftake(n, iterable):

"""只取前 n 个"""

for i, item inenumerate(iterable):

if i >= n:

return

yield item

# 组合：取前5个正数的平方

result = list(take(5, squared(positive_numbers(range(-10, 20)))))

print(result)  # [1, 4, 9, 16, 25]

```

### 练习 3：JSON 配置文件的读写工具类

### 练习 4：用 try-except 实现输入验证循环

```python

whileTrue:

try:

        age = int(input("请输入年龄: "))

if age < 0:

raiseValueError("年龄不能为负数")

print(f"你输入的年龄是: {age}")

break

exceptValueErroras e:

print(f"❌ 输入无效: {e}，请重试")

```

### 练习 5：用 with 语句实现简单的数据库连接池（思考题）

---

## 下期预告

**Python 教程 Episode 05 — 爬虫入门：用 requests + BeautifulSoup 抓网页数据**

- HTTP 基础知识（GET/POST 请求）

- 使用 requests 库发送网络请求

- 解析 HTML：BeautifulSoup 入门

- 反爬策略应对（headers、session、代理）

- 实战：爬取豆瓣电影 Top250

- 进阶：Playwright 抓取动态渲染页面（AJAX/JS 渲染）

---

*如果觉得本系列对你有帮助，欢迎点赞、评论、转发！有任何问题，在留言区讨论。*