章节5 高级特性与异常处理

授课时长：约5-6学时
适用对象：已掌握面向对象编程的学员
教学方式：讲授 + 代码演示 + 调试实战

一、学习目标

目标分类	具体目标
知识目标	理解文件IO与上下文管理器、异常机制、操作系统接口、并发编程模型、单元测试方法论
技能目标	熟练使用 pathlib 处理路径、编写线程/进程/协程程序、用 pytest 编写测试用例
素养目标	培养鲁棒编程意识，写出"防御性"代码，建立测试驱动开发（TDD）思维

二、核心知识点

1️⃣ 文件IO操作与上下文管理器

1.1 文件操作基础

python

"""
文件操作的三个阶段：
1. 打开文件 → open()
2. 读写文件 → read()/write()/readline()/readlines()
3. 关闭文件 → close()
"""

# ──────────────────────────────────────────
# 打开模式详解
"""
'r'   读取（默认）
'w'   写入（覆盖已有内容）
'a'   追加（在末尾写入）
'x'   排他创建（文件已存在则失败）
'b'   二进制模式（与上述组合：'rb', 'wb'）
't'   文本模式（默认）
'+'   读写模式（'r+', 'w+'）
"""

# ──────────────────────────────────────────
# 方式1：传统方式（需手动关闭）
f = open("example.txt", "w", encoding="utf-8")
f.write("Hello, World!\n")
f.write("这是第二行\n")
f.close()

f = open("example.txt", "r", encoding="utf-8")
content = f.read()
print(content)   # 读取全部内容
f.close()

# ──────────────────────────────────────────
# 方式2：使用 with 语句（推荐 ✅）
with open("example.txt", "r", encoding="utf-8") as f:
    content = f.read()
    print(content)
# 文件自动关闭，无需手动 close()

1.2 常用读写方法

python

# ──────────────────────────────────────────
# 读取方法
with open("example.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n第二行\n第三行\n第四行\n第五行\n")

# read() — 读取全部
with open("example.txt", "r", encoding="utf-8") as f:
    content = f.read()
    print(repr(content))        # '第一行\n第二行\n第三行\n第四行\n第五行\n'

# readline() — 逐行读取
with open("example.txt", "r", encoding="utf-8") as f:
    line = f.readline()          # '第一行\n'
    print(f"第一行: {line.strip()}")

# readlines() — 读取所有行到列表
with open("example.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()
    print(f"共{len(lines)}行")

# 最佳实践：直接遍历文件对象（逐行处理，内存友好）
with open("example.txt", "r", encoding="utf-8") as f:
    for line in f:               # ✅ 逐行读取，不一次性加载全部
        print(line.strip())

# ──────────────────────────────────────────
# 写入方法
# write() — 写入字符串
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello\n")
    f.write("World\n")

# writelines() — 写入字符串列表
lines = ["第一行\n", "第二行\n", "第三行\n"]
with open("output.txt", "a", encoding="utf-8") as f:
    f.writelines(lines)          # 追加写入

# ──────────────────────────────────────────
# 文件指针操作
with open("example.txt", "r", encoding="utf-8") as f:
    print(f.tell())              # 0（当前指针位置）
    f.read(3)                    # 读取3个字符
    print(f.tell())              # 3
    f.seek(0)                    # 回到文件开头
    print(f.tell())              # 0

1.3 二进制文件操作

python

# 操作图片、视频等二进制文件
with open("example.jpg", "rb") as f:
    data = f.read(1024)          # 读取前1024字节
    print(f"读取了{len(data)}字节")

# 复制文件
with open("source.jpg", "rb") as src:
    with open("dest.jpg", "wb") as dst:
        while chunk := src.read(8192):  # 海象运算符 :=
            dst.write(chunk)

1.4 上下文管理器（深入）

python

# ──────────────────────────────────────────
# 方式1：基于类的上下文管理器
class Timer:
    """计时器上下文管理器"""
    def __enter__(self):
        import time
        self.start = time.time()
        return self  # 返回给 as 子句

    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        elapsed = time.time() - self.start
        print(f"耗时: {elapsed:.4f}秒")
        return False  # False = 不吞异常, True = 吞异常

# 使用
with Timer():
    total = sum(range(10000000))
    print(f"总和: {total}")

# ──────────────────────────────────────────
# 方式2：使用 contextlib
from contextlib import contextmanager

@contextmanager
def managed_resource(*args, **kwargs):
    """使用生成器实现的上下文管理器"""
    # __enter__: 创建资源
    resource = {"name": "my_resource"}
    print("▶ 获取资源")
    try:
        yield resource  # 将资源返回给 with 块
    finally:
        # __exit__: 清理资源
        print("◀ 释放资源")

# 使用
with managed_resource() as res:
    print(f"使用中: {res}")

# ──────────────────────────────────────────
# 多个上下文管理器
with open("a.txt", "w") as f1, open("b.txt", "w") as f2:
    f1.write("file A")
    f2.write("file B")

2️⃣ 异常捕获与自定义异常

2.1 异常处理基础

python

# ──────────────────────────────────────────
# try-except 基本结构
try:
    num = int(input("请输入数字: "))
    result = 10 / num
    print(f"结果: {result}")
except ValueError:
    print("❌ 输入必须是数字")
except ZeroDivisionError:
    print("❌ 不能除以0")
except Exception as e:          # 捕获所有其他异常
    print(f"❌ 未知错误: {e}")
else:
    print("✅ 没有发生异常")
finally:
    print("🔚 无论是否异常都会执行")

2.2 捕获多个异常

python

# 方式1：多个 except 块
try:
    data = [1, 2, 3]
    index = int(input("输入索引: "))
    print(data[index] / 0)
except (IndexError, ZeroDivisionError) as e:
    print(f"索引或除零错误: {e}")
except ValueError:
    print("输入不是有效数字")

# 方式2：捕获并重新抛出（保留堆栈信息）
import traceback

try:
    1 / 0
except ZeroDivisionError:
    print("记录日志...")
    traceback.print_exc()        # 打印完整堆栈
    # raise                    # 重新抛出原异常（不丢失堆栈）

2.3 自定义异常

python

# ──────────────────────────────────────────
# 继承 Exception 自定义异常
class InsufficientBalanceError(Exception):
    """余额不足异常"""
    def __init__(self, balance: float, amount: float):
        self.balance = balance
        self.amount = amount
        self.shortfall = amount - balance
        super().__init__(f"余额不足！当前余额: {balance}元，需要: {amount}元，差额: {self.shortfall}元")


class NegativeAmountError(Exception):
    """金额不能为负数"""
    pass


class BankAccount:
    def __init__(self, owner: str, balance: float = 0):
        self.owner = owner
        self.balance = balance

    def withdraw(self, amount: float):
        if amount < 0:
            raise NegativeAmountError("取款金额不能为负数")
        if amount > self.balance:
            raise InsufficientBalanceError(self.balance, amount)
        self.balance -= amount
        return f"取款 {amount}元成功，余额: {self.balance}元"

    def deposit(self, amount: float):
        if amount < 0:
            raise NegativeAmountError("存款金额不能为负数")
        self.balance += amount
        return f"存款 {amount}元成功，余额: {self.balance}元"


# 使用
acc = BankAccount("张三", 1000)

try:
    acc.withdraw(2000)
except InsufficientBalanceError as e:
    print(f"❌ {e}")
    print(f"差额: {e.shortfall}元")
except NegativeAmountError as e:
    print(f"❌ {e}")
except Exception as e:
    print(f"❌ 未知错误: {e}")
else:
    print("✅ 操作成功")

2.4 异常处理最佳实践

python

"""
异常处理的核心原则：
1. 只在真正需要异常处理的地方使用 try，不要用异常控制流程
2. 捕获异常要具体，不要用 except: pass 静默所有异常
3. 异常信息要有意义，方便排查问题
4. 资源清理放 finally 或使用 with 语句
"""

# ❌ 坏实践：用异常控制流程
def get_value_bad(d, key):
    try:
        return d[key]
    except KeyError:
        return None

# ✅ 好实践：用条件判断
def get_value_good(d, key):
    return d.get(key)

# ❌ 坏实践：空的 except
try:
    risky_operation()
except:           # 连 Exception 都不写
    pass          # 所有错误都被吞掉

# ✅ 好实践：至少记录日志
import logging
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

try:
    risky_operation()
except Exception as e:
    logger.error(f"操作失败: {e}", exc_info=True)
    raise        # 如果不确定如何处理，重新抛出

3️⃣ OS系统模块与路径处理

3.1 os 模块常用功能

python

import os

# ──────────────────────────────────────────
# 文件和目录操作
print(os.getcwd())             # 当前工作目录
# os.chdir("/path/to/dir")     # 切换目录
# os.mkdir("new_folder")       # 创建单个目录
# os.makedirs("a/b/c")         # 创建多级目录
# os.rmdir("empty_folder")     # 删除空目录
# os.remove("file.txt")        # 删除文件
# os.rename("old.txt", "new.txt")  # 重命名
# os.listdir(".")              # 列出目录内容

# ──────────────────────────────────────────
# 路径操作（os.path）
print(os.path.abspath("."))        # 绝对路径
print(os.path.basename("/a/b/c.txt"))  # 'c.txt'
print(os.path.dirname("/a/b/c.txt"))    # '/a/b'
print(os.path.split("/a/b/c.txt"))      # ('/a/b', 'c.txt')
print(os.path.splitext("file.txt"))     # ('file', '.txt')
print(os.path.exists("test.txt"))       # 是否存在
print(os.path.isfile("test.txt"))       # 是否是文件
print(os.path.isdir("test.txt"))        # 是否是目录
print(os.path.getsize("test.txt"))      # 文件大小（字节）

# ──────────────────────────────────────────
# 环境变量
print(os.environ.get("PATH"))       # 获取环境变量
print(os.environ.get("HOME"))       # 用户主目录

# ──────────────────────────────────────────
# 系统信息
print(os.name)                     # 'nt' / 'posix'
print(os.cpu_count())              # CPU核心数
print(os.getpid())                 # 当前进程ID

3.2 pathlib 模块（Python 3.4+）

pathlib 是 Python 推荐的面向对象路径处理方式，统一了不同操作系统的路径分隔符。

python

from pathlib import Path

# ──────────────────────────────────────────
# 创建路径
p = Path(".")                          # 当前目录
home = Path.home()                     # 用户主目录
cwd = Path.cwd()                       # 当前工作目录

# ──────────────────────────────────────────
# 路径组合
data_dir = Path("data") / "images" / "2024"
print(data_dir)                        # data\images\2024

# ──────────────────────────────────────────
# 路径属性
path = Path("/home/user/docs/file.txt")
print(path.name)         # 'file.txt'
print(path.stem)         # 'file'（无扩展名）
print(path.suffix)       # '.txt'
print(path.parent)       # /home/user/docs
print(path.parents[0])   # /home/user/docs
print(path.parents[1])   # /home/user
print(path.anchor)       # '/'（根目录）

# ──────────────────────────────────────────
# 文件操作
path = Path("test.txt")

# 读写（一行搞定！）
path.write_text("Hello, pathlib!", encoding="utf-8")
content = path.read_text(encoding="utf-8")
print(content)                      # Hello, pathlib!

# 二进制读写
path.write_bytes(b"binary data")
data = path.read_bytes()

# ──────────────────────────────────────────
# 目录遍历
path = Path(".")

# 遍历所有 .py 文件
for py_file in path.glob("**/*.py"):
    print(py_file)

# 遍历当前目录下的所有文件
for item in path.iterdir():
    if item.is_file():
        print(f"📄 {item.name} ({item.stat().st_size} bytes)")
    elif item.is_dir():
        print(f"📁 {item.name}/")

# ──────────────────────────────────────────
# 创建/删除目录
Path("new_dir/sub_dir").mkdir(parents=True, exist_ok=True)
# Path("new_dir").rmdir()                # 删除空目录

# ──────────────────────────────────────────
# 文件信息
path = Path("test.txt")
if path.exists():
    stat = path.stat()
    print(f"大小: {stat.st_size} bytes")
    print(f"修改时间: {stat.st_mtime}")
    print(f"权限: {stat.st_mode:o}")

# ──────────────────────────────────────────
# 实际应用：批量重命名
def batch_rename(directory: str, old_suffix: str, new_suffix: str):
    """批量修改文件扩展名"""
    path = Path(directory)
    for file in path.glob(f"*{old_suffix}"):
        new_name = file.with_suffix(new_suffix)
        file.rename(new_name)
        print(f"重命名: {file.name} → {new_name.name}")

# 使用
# batch_rename("./docs", ".txt", ".md")

# ──────────────────────────────────────────
# 路径判断
p = Path("example.txt")
print(p.exists())         # 是否存在
print(p.is_file())        # 是否是文件
print(p.is_dir())         # 是否是目录
print(p.is_absolute())    # 是否是绝对路径
print(p.resolve())        # 解析为绝对路径

3.3 os.path vs pathlib 对比

操作	os.path	pathlib
拼接路径	`os.path.join("a", "b")`	`Path("a") / "b"`
文件名	`os.path.basename(p)`	`Path(p).name`
扩展名	`os.path.splitext(p)[1]`	`Path(p).suffix`
是否存在	`os.path.exists(p)`	`Path(p).exists()`
读写文本	`open(p).read()`	`Path(p).read_text()`
遍历文件	`os.listdir()`	`path.glob()`

建议：新项目统一使用 pathlib，更简洁、更Pythonic。

4️⃣ 多线程多进程与协程入门

4.1 多线程（threading）

python

"""
多线程：适合 I/O 密集型任务（文件读写、网络请求、数据库操作）
Python GIL（全局解释器锁）使多线程不能并行执行 CPU 密集型任务
"""

import threading
import time
from typing import List


# ──────────────────────────────────────────
# 方式1：使用 Thread 类
def worker(name: str, delay: float):
    """模拟耗时任务"""
    print(f"▶ 线程 {name} 启动")
    time.sleep(delay)
    print(f"◀ 线程 {name} 完成")


# 创建线程
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(f"T{i+1}", i + 1))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("所有线程执行完毕")

# ──────────────────────────────────────────
# 方式2：继承 Thread 类
class DownloadThread(threading.Thread):
    """下载线程"""

    def __init__(self, url: str):
        super().__init__()
        self.url = url
        self.result: str = ""

    def run(self):
        """线程执行体"""
        print(f"下载: {self.url}")
        time.sleep(1)  # 模拟下载
        self.result = f"{self.url} 下载完成"


# 使用
urls = ["http://a.com", "http://b.com", "http://c.com"]
downloaders = [DownloadThread(url) for url in urls]

for d in downloaders:
    d.start()

for d in downloaders:
    d.join()
    print(d.result)

# ──────────────────────────────────────────
# 线程安全与锁
counter = 0
lock = threading.Lock()

def safe_increment():
    global counter
    for _ in range(100000):
        with lock:          # 使用锁保护共享资源
            counter += 1

threads = [threading.Thread(target=safe_increment) for _ in range(10)]
[t.start() for t in threads]
[t.join() for t in threads]
print(f"计数器结果: {counter}")  # 1000000（正确）

4.2 多进程（multiprocessing）

python

"""
多进程：适合 CPU 密集型任务（计算、图像处理、数据分析）
每个进程有独立的 Python 解释器和内存空间，不受 GIL 限制
"""

import multiprocessing
import time


# ──────────────────────────────────────────
# 基本使用
def cpu_heavy_calc(n: int) -> int:
    """CPU密集型计算"""
    total = 0
    for i in range(n):
        total += i ** 2
    return total

if __name__ == "__main__":
    # 串行执行
    start = time.time()
    results = [cpu_heavy_calc(5000000) for _ in range(4)]
    print(f"串行耗时: {time.time() - start:.2f}秒")

    # 并行执行
    start = time.time()
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(cpu_heavy_calc, [5000000] * 4)
    print(f"并行耗时: {time.time() - start:.2f}秒")

# ──────────────────────────────────────────
# 进程间通信（Queue）
def producer(queue):
    for i in range(5):
        queue.put(f"消息 {i}")
        time.sleep(0.1)

def consumer(queue):
    while True:
        msg = queue.get()
        if msg == "DONE":
            break
        print(f"收到: {msg}")

if __name__ == "__main__":
    queue = multiprocessing.Queue()

    p = multiprocessing.Process(target=producer, args=(queue,))
    c = multiprocessing.Process(target=consumer, args=(queue,))

    p.start()
    c.start()

    p.join()
    queue.put("DONE")
    c.join()

4.3 线程池与进程池

python

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time


# ──────────────────────────────────────────
# 线程池（I/O 密集型）
def fetch_url(url: str) -> str:
    time.sleep(0.5)  # 模拟网络请求
    return f"✅ {url} 响应成功"

urls = [f"http://site.com/page/{i}" for i in range(10)]

start = time.time()
with ThreadPoolExecutor(max_workers=5) as executor:
    # 方式1：逐个提交
    futures = [executor.submit(fetch_url, url) for url in urls]
    for future in futures:
        print(future.result())

# 方式2：map 批量提交
# results = executor.map(fetch_url, urls)

print(f"线程池耗时: {time.time() - start:.2f}秒")

# ──────────────────────────────────────────
# 进程池（CPU 密集型）
def square(n: int) -> int:
    """计算平方（CPU密集型）"""
    return sum(i * i for i in range(n))

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(square, [1000000] * 8))
    print(f"计算结果: {len(results)}个")

4.4 协程与异步IO入门（async/await）

python

"""
协程：单线程内的并发，适合 I/O 密集型任务。
通过 async/await 关键字实现协作式多任务。
"""

import asyncio
import time


# ──────────────────────────────────────────
# 定义协程
async def hello(name: str, delay: float):
    """定义一个协程函数"""
    print(f"▶ 你好 {name}")
    await asyncio.sleep(delay)  # 模拟 I/O 等待
    print(f"◀ 再见 {name}")
    return f"完成: {name}"

# 运行单个协程
# result = asyncio.run(hello("Alice", 1))
# print(result)

# ──────────────────────────────────────────
# 并发运行多个协程
async def main():
    """主协程"""
    start = time.time()

    # 方式1：顺序执行
    # await hello("A", 2)
    # await hello("B", 2)
    # await hello("C", 2)
    # 总耗时约6秒

    # 方式2：并发执行 ✅
    results = await asyncio.gather(
        hello("A", 2),
        hello("B", 2),
        hello("C", 2),
    )
    # 总耗时约2秒

    print(f"并发耗时: {time.time() - start:.2f}秒")
    print(f"结果: {results}")

# asyncio.run(main())

# ──────────────────────────────────────────
# 异步上下文管理器
class AsyncResource:
    async def __aenter__(self):
        print("▶ 获取异步资源")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("◀ 释放异步资源")
        await asyncio.sleep(0.1)

    async def use(self):
        await asyncio.sleep(0.2)
        print("使用中...")

async def demo_async_context():
    async with AsyncResource() as res:
        await res.use()

# asyncio.run(demo_async_context())

# ──────────────────────────────────────────
# 异步 HTTP 请求示例
async def fetch_data(url: str):
    """异步获取数据"""
    # 需要安装 aiohttp: pip install aiohttp
    # async with aiohttp.ClientSession() as session:
    #     async with session.get(url) as response:
    #         return await response.text()
    await asyncio.sleep(0.5)
    return f"数据来自 {url}"

async def main_fetch():
    urls = ["http://api1.com", "http://api2.com", "http://api3.com"]
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

5️⃣ 代码重构与单元测试基础

5.1 代码重构原则

python

"""
重构：在不改变外部行为的前提下，改善代码内部结构。

核心原则（摘自《重构》—— Martin Fowler）：
1. 提取函数（Extract Function）
2. 内联函数（Inline Function）
3. 提取变量（Extract Variable）
4. 移动语句（Slide Statements）
5. 拆分循环（Split Loop）
6. 用多态替换条件（Replace Conditional with Polymorphism）
"""

# ──────────────────────────────────────────
# 示例：重构前（糟糕代码）
def process_order_bad(order):
    total = 0
    for item in order["items"]:
        total += item["price"] * item["quantity"]
    if total > 1000:
        total *= 0.9
    if order["country"] == "CN":
        total += total * 0.13  # 增值税
    elif order["country"] == "US":
        total += total * 0.08
    # ... 更多业务逻辑
    return total

# ──────────────────────────────────────────
# 示例：重构后（清晰代码）
def calculate_subtotal(items) -> float:
    """计算商品小计"""
    return sum(item["price"] * item["quantity"] for item in items)

def apply_discount(total: float) -> float:
    """应用折扣"""
    return total * 0.9 if total > 1000 else total

def apply_tax(total: float, country: str) -> float:
    """应用税率"""
    tax_rates = {"CN": 0.13, "US": 0.08, "JP": 0.10}
    rate = tax_rates.get(country, 0)
    return total * (1 + rate)

def process_order_good(order) -> float:
    """处理订单（重构版）"""
    total = calculate_subtotal(order["items"])
    total = apply_discount(total)
    total = apply_tax(total, order["country"])
    return total

5.2 单元测试基础（pytest）

python

"""
pytest 是目前最流行的 Python 测试框架。
安装：pip install pytest
运行：pytest test_xxx.py
"""

# ──────────────────────────────────────────
# 被测试代码（calculator.py）
class Calculator:
    """计算器类"""

    def add(self, a: float, b: float) -> float:
        return a + b

    def subtract(self, a: float, b: float) -> float:
        return a - b

    def multiply(self, a: float, b: float) -> float:
        return a * b

    def divide(self, a: float, b: float) -> float:
        if b == 0:
            raise ValueError("不能除以0")
        return a / b


# ──────────────────────────────────────────
# 测试代码（test_calculator.py）
"""
import pytest
from calculator import Calculator


# 测试函数命名以 test_ 开头
def test_add():
    calc = Calculator()
    assert calc.add(2, 3) == 5
    assert calc.add(-1, 1) == 0
    assert calc.add(0, 0) == 0


def test_subtract():
    calc = Calculator()
    assert calc.subtract(5, 3) == 2
    assert calc.subtract(0, 5) == -5


def test_multiply():
    calc = Calculator()
    assert calc.multiply(2, 3) == 6
    assert calc.multiply(0, 5) == 0
    assert calc.multiply(-2, 3) == -6


def test_divide():
    calc = Calculator()
    assert calc.divide(10, 2) == 5
    assert calc.divide(7, 2) == 3.5


def test_divide_by_zero():
    calc = Calculator()
    with pytest.raises(ValueError, match="不能除以0"):
        calc.divide(10, 0)
"""

# ──────────────────────────────────────────
# pytest 高级特性

# 1. 参数化测试（parametrize）
"""
import pytest

@pytest.mark.parametrize("a,b,expected", [
    (2, 3, 5),
    (-1, 1, 0),
    (0, 0, 0),
    (100, 200, 300),
])
def test_add_parametrized(a, b, expected):
    calc = Calculator()
    assert calc.add(a, b) == expected
"""

# 2. Fixture（夹具：测试前的准备和清理）
"""
import pytest
from calculator import Calculator

@pytest.fixture
def calc():
    # Setup
    calculator = Calculator()
    yield calculator
    # Teardown（清理）
    print("测试结束，清理资源")


def test_add_with_fixture(calc):
    assert calc.add(2, 3) == 5


def test_subtract_with_fixture(calc):
    assert calc.subtract(5, 3) == 2
"""

# 3. 测试覆盖率
"""
# 安装：pip install pytest-cov
# 运行：pytest --cov=calculator test_calculator.py
# 生成报告：pytest --cov=calculator --cov-report=html test_calculator.py
"""

5.3 pytest 实战示例

python

"""
完整的测试示例：

# 被测代码：user_manager.py
"""

class UserManager:
    """用户管理器"""

    def __init__(self):
        self._users = {}

    def register(self, username: str, email: str) -> bool:
        """注册新用户"""
        if username in self._users:
            return False
        if "@" not in email:
            raise ValueError("无效的邮箱格式")
        self._users[username] = {"username": username, "email": email}
        return True

    def get_user(self, username: str) -> dict:
        """获取用户信息"""
        if username not in self._users:
            raise KeyError(f"用户 {username} 不存在")
        return self._users[username]

    def delete_user(self, username: str) -> bool:
        """删除用户"""
        return self._users.pop(username, None) is not None

    def count(self) -> int:
        """用户数量"""
        return len(self._users)


"""
# 测试代码：test_user_manager.py
import pytest
from user_manager import UserManager


@pytest.fixture
def manager():
    """每个测试函数都会使用新的 UserManager 实例"""
    return UserManager()


class TestUserManager:
    """测试类（按功能分组）"""

    def test_register_success(self, manager):
        assert manager.register("alice", "alice@test.com") is True
        assert manager.count() == 1

    def test_register_duplicate(self, manager):
        manager.register("alice", "alice@test.com")
        assert manager.register("alice", "alice@test.com") is False

    def test_register_invalid_email(self, manager):
        with pytest.raises(ValueError, match="无效的邮箱格式"):
            manager.register("bob", "invalid-email")

    def test_get_user(self, manager):
        manager.register("alice", "alice@test.com")
        user = manager.get_user("alice")
        assert user["username"] == "alice"
        assert user["email"] == "alice@test.com"

    def test_get_nonexistent_user(self, manager):
        with pytest.raises(KeyError):
            manager.get_user("nonexistent")

    def test_delete_user(self, manager):
        manager.register("alice", "alice@test.com")
        assert manager.delete_user("alice") is True
        assert manager.count() == 0

    def test_delete_nonexistent_user(self, manager):
        assert manager.delete_user("nonexistent") is False
"""

# 运行测试命令
# pytest test_user_manager.py -v              # 详细输出
# pytest test_user_manager.py::TestUserManager::test_register_success  # 运行单个测试
# pytest -x test_user_manager.py              # 遇到第一个失败就停止
# pytest --tb=short test_user_manager.py      # 简化堆栈信息

三、代码实操演示

演示1：日志文件分析工具

python

"""综合运用文件IO、异常处理、pathlib、正则表达式"""
import re
from pathlib import Path
from collections import defaultdict


class LogAnalyzer:
    """日志文件分析器"""

    def __init__(self, log_dir: str):
        self.log_dir = Path(log_dir)
        if not self.log_dir.exists():
            raise FileNotFoundError(f"目录不存在: {log_dir}")

    def search_errors(self, pattern: str = "ERROR") -> list:
        """搜索错误日志"""
        errors = []
        for log_file in self.log_dir.glob("*.log"):
            try:
                content = log_file.read_text(encoding="utf-8", errors="ignore")
                for line_num, line in enumerate(content.splitlines(), 1):
                    if pattern in line:
                        errors.append({
                            "file": log_file.name,
                            "line": line_num,
                            "content": line.strip()
                        })
            except PermissionError:
                print(f"⚠️ 无权限读取: {log_file}")
            except Exception as e:
                print(f"⚠️ 读取 {log_file} 时出错: {e}")
        return errors

    def count_log_levels(self) -> dict:
        """统计不同日志级别的数量"""
        counts = defaultdict(int)
        pattern = re.compile(r"\b(DEBUG|INFO|WARN|ERROR|FATAL)\b")

        for log_file in self.log_dir.glob("*.log"):
            try:
                for line in log_file.read_text(
                    encoding="utf-8", errors="ignore"
                ).splitlines():
                    for match in pattern.finditer(line):
                        counts[match.group(1)] += 1
            except Exception:
                continue
        return dict(counts)

    def generate_report(self, output_path: str):
        """生成分析报告"""
        errors = self.search_errors()
        levels = self.count_log_levels()

        report = []
        report.append("=" * 50)
        report.append("日志分析报告")
        report.append("=" * 50)
        report.append(f"\n📊 日志级别统计:")
        report.append("-" * 20)
        for level, count in sorted(levels.items()):
            bar = "█" * min(count, 50)
            report.append(f"  {level:5s}: {count:5d} {bar}")

        report.append(f"\n❌ 错误详情 (共{len(errors)}条):")
        report.append("-" * 30)
        for e in errors[:20]:  # 最多显示20条
            report.append(f"  [{e['file']}:{e['line']}] {e['content']}")

        Path(output_path).write_text(
            "\n".join(report), encoding="utf-8"
        )
        print(f"报告已生成: {output_path}")


# 使用
if __name__ == "__main__":
    analyzer = LogAnalyzer("./logs")
    analyzer.generate_report("analysis_report.txt")

演示2：并发下载器

python

"""多线程/协程并发下载器"""

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path


class Downloader:
    """并发下载器"""

    def __init__(self, output_dir: str = "./downloads"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    # ──────────────────────────────────────────
    # 线程版本（I/O密集型）
    def download_thread(self, url: str) -> str:
        """模拟下载（单文件）"""
        time.sleep(0.5)
        filename = url.split("/")[-1] or f"file_{hash(url)}.txt"
        filepath = self.output_dir / filename
        filepath.write_text(f"Content from {url}", encoding="utf-8")
        return filename

    def batch_download_thread(self, urls: list) -> list:
        """批量下载（线程池）"""
        start = time.time()
        with ThreadPoolExecutor(max_workers=5) as executor:
            results = list(executor.map(self.download_thread, urls))
        print(f"线程池下载 {len(urls)} 个文件，耗时: {time.time() - start:.2f}s")
        return results

    # ──────────────────────────────────────────
    # 协程版本（更轻量）
    async def download_async(self, url: str) -> str:
        """异步下载"""
        await asyncio.sleep(0.5)
        filename = url.split("/")[-1] or f"file_{hash(url)}.txt"
        filepath = self.output_dir / filename
        filepath.write_text(f"Content from {url}", encoding="utf-8")
        return filename

    async def batch_download_async(self, urls: list) -> list:
        """批量下载（协程）"""
        start = time.time()
        tasks = [self.download_async(url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"协程下载 {len(urls)} 个文件，耗时: {time.time() - start:.2f}s")
        return results


# 使用
if __name__ == "__main__":
    urls = [f"http://example.com/file_{i}.txt" for i in range(10)]

    downloader = Downloader()

    # 线程版本
    downloader.batch_download_thread(urls)

    # 协程版本
    # asyncio.run(downloader.batch_download_async(urls))

四、常见错误与排错

错误类型	示例	原因与解决
FileNotFoundError	`open("no_file.txt")`	文件不存在，检查路径或使用 `path.exists()` 预检
PermissionError	没有读取/写入权限	检查文件权限，或以管理员身份运行
UnicodeEncodeError	写入中文时报错	指定 `encoding="utf-8"`
OSError: [Errno 22]	路径包含非法字符	检查路径中的特殊字符，Windows路径中的 `\` 要转义
TimeoutError	线程/网络操作超时	设置合理的超时时间，或使用 async 超时机制
RuntimeError: asyncio.run() cannot be called from a running event loop	在 Jupyter 等环境中使用 `asyncio.run()`	改用 `await` 或 `nest_asyncio.apply()`
AssertionError	`assert result == expected` 失败	检查测试预期值是否正确，或代码逻辑是否有bug

五、小结

知识图谱

高级特性与异常处理
├── 文件IO
│   ├── 模式：r/w/a/x/b/t/+
│   ├── 方法：read/write/readline/readlines
│   ├── 指针：tell()/seek()
│   └── with语句：__enter__/__exit__/contextlib
├── 异常处理
│   ├── try-except-else-finally
│   ├── 自定义异常：继承 Exception
│   └── 最佳实践：精确捕获、记录日志
├── 路径处理
│   ├── os.path：传统方式
│   └── pathlib：面向对象方式（推荐 ✅）
├── 并发编程
│   ├── 多线程（threading）：I/O密集型
│   ├── 多进程（multiprocessing）：CPU密集型
│   ├── 进程/线程池（concurrent.futures）
│   └── 协程（async/await）：轻量并发
└── 测试与重构
    ├── 重构原则：提取函数/变量、拆分
    └── pytest：fixture/parametrize/覆盖率

重点回顾

with 语句是文件操作的标准方式，自动管理资源释放
pathlib 比 os.path 更简洁、更跨平台，新项目优先使用
并发模型选择：I/O密集型用多线程或协程，CPU密集型用多进程
异常处理的核心是"精确捕获"——不要用 except: 吞掉所有异常
测试先行（TDD）：先写测试再写代码，提高代码质量和可维护性

六、课后练习

基础题

文件操作：编写程序，读取一个文本文件，统计其中包含多少个单词、多少行、多少个字符（含空格和不含空格）。
异常处理：编写一个除法计算器，处理除零错误和输入非数字错误，要求循环运行直到用户输入"exit"。
路径处理：使用 pathlib 遍历指定目录，列出所有 .txt 文件及其大小（格式化为KB/MB），按大小从大到小排序。

进阶题

多线程爬虫：编写一个多线程程序，并发下载5个网页内容并保存到本地文件，统计总耗时并与串行版本对比。
自定义异常：设计一个 BankSystem，包含 WithdrawError（取款异常，含余额、请求金额、差额等属性）和 TransferError（转账异常），在业务逻辑中合理使用。
pytest 测试：为你在章节4中设计的 BankAccount 类编写完整的 pytest 测试用例，覆盖正常场景和异常场景（至少6个测试函数）。

挑战题

异步文件搜索器：使用 asyncio + aiofiles 实现一个异步文件内容搜索器，在指定目录中递归搜索包含关键字的文件，返回匹配的文件路径和行号。对比同功能的同步版本性能差异。
完整测试套件：为之前设计的"图书管理系统"（章节4）编写完整的 pytest 测试套件，包含 fixture、参数化测试、异常测试，并生成 HTML 覆盖率报告。

参考资料
Python 官方文档-文件IO：https://docs.python.org/zh-cn/3/tutorial/inputoutput.html
Python 官方文档-异常：https://docs.python.org/zh-cn/3/tutorial/errors.html
pathlib 文档：https://docs.python.org/zh-cn/3/library/pathlib.html
asyncio 文档：https://docs.python.org/zh-cn/3/library/asyncio.html
pytest 官方文档：https://docs.pytest.org/en/stable/

章节5 高级特性与异常处理 ​

一、学习目标 ​

二、核心知识点 ​

1️⃣ 文件IO操作与上下文管理器 ​

1.1 文件操作基础 ​

1.2 常用读写方法 ​

1.3 二进制文件操作 ​

1.4 上下文管理器（深入） ​

2️⃣ 异常捕获与自定义异常 ​

2.1 异常处理基础 ​

2.2 捕获多个异常 ​

2.3 自定义异常 ​

2.4 异常处理最佳实践 ​

3️⃣ OS系统模块与路径处理 ​

3.1 os 模块常用功能 ​

3.2 pathlib 模块（Python 3.4+） ​

3.3 os.path vs pathlib 对比 ​

4️⃣ 多线程多进程与协程入门 ​

4.1 多线程（threading） ​

4.2 多进程（multiprocessing） ​

4.3 线程池与进程池 ​

4.4 协程与异步IO入门（async/await） ​

5️⃣ 代码重构与单元测试基础 ​

5.1 代码重构原则 ​

5.2 单元测试基础（pytest） ​

5.3 pytest 实战示例 ​

三、代码实操演示 ​

演示1：日志文件分析工具 ​

演示2：并发下载器 ​

四、常见错误与排错 ​

五、小结 ​

知识图谱 ​

重点回顾 ​

六、课后练习 ​

基础题 ​

进阶题 ​

挑战题 ​

章节5 高级特性与异常处理

一、学习目标

二、核心知识点

1️⃣ 文件IO操作与上下文管理器

1.1 文件操作基础

1.2 常用读写方法

1.3 二进制文件操作

1.4 上下文管理器（深入）

2️⃣ 异常捕获与自定义异常

2.1 异常处理基础

2.2 捕获多个异常

2.3 自定义异常

2.4 异常处理最佳实践

3️⃣ OS系统模块与路径处理

3.1 os 模块常用功能

3.2 pathlib 模块（Python 3.4+）

3.3 os.path vs pathlib 对比

4️⃣ 多线程多进程与协程入门

4.1 多线程（threading）

4.2 多进程（multiprocessing）

4.3 线程池与进程池

4.4 协程与异步IO入门（async/await）

5️⃣ 代码重构与单元测试基础

5.1 代码重构原则

5.2 单元测试基础（pytest）

5.3 pytest 实战示例

三、代码实操演示

演示1：日志文件分析工具

演示2：并发下载器

四、常见错误与排错

五、小结

知识图谱

重点回顾

六、课后练习

基础题

进阶题

挑战题