老周有AI

主题

阅读: 0

🎯 上下文管理器练习：资源管理的魔法守护 ​

🎯 练习目标 ​

通过本次练习，你将：

  • 掌握 Python 上下文管理器的基本概念和工作原理
  • 学会使用 with 语句管理资源
  • 能够使用类和 contextlib 模块创建自定义上下文管理器
  • 了解上下文管理器在实际开发中的应用场景

📝 基础练习 ​

练习 1：文件操作的上下文管理 ​

任务：编写一个程序，使用 with 语句打开一个文本文件，读取其中的内容，并统计文件中的单词数量。

提示：

  • 使用 with open() 语句确保文件正确关闭
  • 可以使用 split() 方法分割文本为单词
  • 处理可能的文件不存在异常

参考实现：

python
# 文件单词计数器
file_path = "sample.txt"

try:
    with open(file_path, 'r', encoding='utf-8') as file:
        content = file.read()
        words = content.split()
        print(f"文件 '{file_path}' 中共有 {len(words)} 个单词。")
except FileNotFoundError:
    print(f"错误：找不到文件 '{file_path}'。")
except Exception as e:
    print(f"发生错误：{e}")

练习 2：自定义计时器上下文管理器 ​

任务：创建一个自定义上下文管理器，用于测量代码块的执行时间。

要求：

  • 使用类的方式实现 __enter__ 和 __exit__ 方法
  • 在 __enter__ 中记录开始时间
  • 在 __exit__ 中计算并输出执行时间
  • 确保异常情况下也能正确工作

参考实现：

python
import time

class Timer:
    def __enter__(self):
        self.start_time = time.time()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        end_time = time.time()
        execution_time = end_time - self.start_time
        print(f"代码执行时间：{execution_time:.4f} 秒")
        # 返回 False 表示不抑制异常
        return False

# 使用示例
with Timer():
    # 模拟耗时操作
    total = sum(i for i in range(1000000))
    print(f"计算结果：{total}")

练习 3：使用 contextlib 简化上下文管理器 ​

任务：使用 contextlib.contextmanager 装饰器重写练习 2 中的计时器上下文管理器。

要求：

  • 使用生成器函数和装饰器实现
  • 功能与练习 2 相同

参考实现：

python
import time
from contextlib import contextmanager

@contextmanager
def timer():
    start_time = time.time()
    try:
        # yield 之前的代码相当于 __enter__
        yield
    finally:
        # yield 之后的代码相当于 __exit__
        end_time = time.time()
        execution_time = end_time - start_time
        print(f"代码执行时间：{execution_time:.4f} 秒")

# 使用示例
with timer():
    # 模拟耗时操作
    total = sum(i for i in range(1000000))
    print(f"计算结果：{total}")

练习 4：资源管理上下文管理器 ​

任务：创建一个上下文管理器，用于安全地管理数据库连接。

要求：

  • 模拟数据库连接的创建和关闭
  • 确保无论是否发生异常，连接都能正确关闭
  • 在 __enter__ 中返回连接对象供 with 语句使用

参考实现：

python
class DatabaseConnection:
    def __init__(self, connection_string):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        print(f"正在连接到数据库：{self.connection_string}")
        # 模拟连接数据库
        self.connection = {"status": "connected", "connection_id": 12345}
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.connection:
            print("正在关闭数据库连接...")
            # 模拟关闭数据库连接
            self.connection = None
            print("数据库连接已关闭")
        # 不抑制异常
        return False

# 使用示例
with DatabaseConnection("mysql://user:password@localhost:3306/mydb") as conn:
    print(f"使用连接：{conn}")
    # 模拟执行查询
    # cursor = conn.cursor()
    # cursor.execute("SELECT * FROM users")
    print("查询执行完成")

练习 5：更改工作目录上下文管理器 ​

任务：创建一个上下文管理器，用于临时更改当前工作目录，并在离开 with 块时恢复原始工作目录。

提示：

  • 使用 os.getcwd() 获取当前目录
  • 使用 os.chdir() 更改目录
  • 在 __exit__ 中恢复原始目录

参考实现：

python
import os

class ChangeDirectory:
    def __init__(self, new_dir):
        self.new_dir = new_dir
        self.original_dir = None
    
    def __enter__(self):
        self.original_dir = os.getcwd()
        try:
            os.chdir(self.new_dir)
            print(f"当前工作目录已更改为：{os.getcwd()}")
        except Exception as e:
            print(f"更改目录失败：{e}")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.original_dir:
            os.chdir(self.original_dir)
            print(f"已恢复原始工作目录：{os.getcwd()}")
        return False

# 使用示例
with ChangeDirectory("../"):
    # 在新目录中执行操作
    print(f"在新目录中列出文件：")
    try:
        files = os.listdir('.')
        print(files)
    except Exception as e:
        print(f"列出文件失败：{e}")

🚀 挑战练习 ​

挑战 1：事务处理上下文管理器 ​

任务：创建一个用于数据库事务处理的上下文管理器，支持提交和回滚操作。

要求：

  • 实现 Transaction 类，支持事务的开始、提交和回滚
  • 在 with 块中正常退出时提交事务
  • 在发生异常时自动回滚事务
  • 提供自定义的 commit() 和 rollback() 方法，允许手动控制

参考实现：

python
class Transaction:
    def __init__(self, connection):
        self.connection = connection
        self.is_committed = False
        self.is_rolled_back = False
    
    def __enter__(self):
        print("事务开始")
        # 假设 connection 有 begin_transaction 方法
        # self.connection.begin_transaction()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            # 发生异常，自动回滚
            if not self.is_committed and not self.is_rolled_back:
                self.rollback()
            # 返回 False 不抑制异常
            return False
        else:
            # 正常退出，自动提交
            if not self.is_committed and not self.is_rolled_back:
                self.commit()
            return True
    
    def commit(self):
        if not self.is_committed and not self.is_rolled_back:
            print("事务提交")
            # self.connection.commit()
            self.is_committed = True
    
    def rollback(self):
        if not self.is_committed and not self.is_rolled_back:
            print("事务回滚")
            # self.connection.rollback()
            self.is_rolled_back = True

# 使用示例
# 假设我们有一个数据库连接对象
class MockConnection:
    def begin_transaction(self):
        print("模拟开始事务")
    def commit(self):
        print("模拟提交事务")
    def rollback(self):
        print("模拟回滚事务")

conn = MockConnection()

# 正常情况 - 事务会自动提交
print("=== 测试正常事务 ===")
with Transaction(conn):
    # 执行数据库操作
    print("执行数据库操作...")

# 异常情况 - 事务会自动回滚
print("\n=== 测试异常事务 ===")
try:
    with Transaction(conn):
        print("执行数据库操作...")
        # 模拟异常
        raise ValueError("模拟数据库错误")
except ValueError as e:
    print(f"捕获到异常：{e}")

# 手动控制提交/回滚
print("\n=== 测试手动控制 ===")
with Transaction(conn) as tx:
    print("执行数据库操作...")
    # 根据条件决定提交或回滚
    should_commit = True
    if should_commit:
        tx.commit()
    else:
        tx.rollback()

挑战 2：多资源上下文管理器 ​

任务：创建一个可以同时管理多个资源的上下文管理器。

要求：

  • 实现一个 MultiContext 类，接受多个上下文管理器作为参数
  • 按顺序进入每个上下文管理器，逆序退出
  • 确保即使在嵌套的上下文中发生异常，所有资源也能被正确释放
  • 支持通过 as 子句获取所有上下文管理器的返回值

参考实现：

python
class MultiContext:
    def __init__(self, *context_managers):
        self.context_managers = context_managers
        self.entered_contexts = []
    
    def __enter__(self):
        results = []
        try:
            for cm in self.context_managers:
                result = cm.__enter__()
                self.entered_contexts.append(cm)
                results.append(result)
            # 如果只有一个上下文管理器，直接返回其结果
            if len(results) == 1:
                return results[0]
            return results
        except Exception as e:
            # 如果在进入过程中发生异常，清理已进入的上下文
            self._cleanup_contexts()
            raise e
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # 逆序退出上下文管理器
        exceptions = []
        for cm in reversed(self.entered_contexts):
            try:
                cm.__exit__(exc_type, exc_val, exc_tb)
            except Exception as e:
                exceptions.append(e)
        
        # 如果在清理过程中也发生了异常，抛出第一个异常
        if exceptions:
            raise exceptions[0]
        
        # 不抑制原始异常
        return False
    
    def _cleanup_contexts(self):
        # 清理已进入的上下文管理器
        for cm in reversed(self.entered_contexts):
            try:
                cm.__exit__(None, None, None)
            except:
                # 忽略清理过程中的异常
                pass

# 使用示例
# 创建一些简单的上下文管理器
class Resource1:
    def __enter__(self):
        print("Resource1 已打开")
        return "Resource1 对象"
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Resource1 已关闭")
        return False

class Resource2:
    def __enter__(self):
        print("Resource2 已打开")
        return "Resource2 对象"
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Resource2 已关闭")
        return False

# 同时管理多个资源
print("=== 测试多资源管理 ===")
with MultiContext(Resource1(), Resource2()) as results:
    res1, res2 = results
    print(f"获得资源: {res1}, {res2}")
    print("使用资源进行操作...")

# 测试异常情况
print("\n=== 测试异常情况 ===")
try:
    with MultiContext(Resource1(), Resource2()) as results:
        res1, res2 = results
        print("使用资源进行操作...")
        # 模拟异常
        raise ValueError("操作出错")
except ValueError as e:
    print(f"捕获到异常: {e}")

💡 实践应用提示 ​

  1. 数据库操作：上下文管理器是处理数据库连接的理想选择，确保连接在使用后被正确关闭。

  2. 文件处理：使用 with open() 是 Python 中处理文件的推荐方式，可以避免忘记关闭文件导致的资源泄露。

  3. 网络连接：创建上下文管理器来管理网络连接、套接字等资源，确保连接在使用后被关闭。

  4. 测试环境：在测试代码中使用上下文管理器来设置和清理测试环境。

  5. 线程锁：可以使用上下文管理器来简化线程锁的获取和释放，避免死锁。

  6. 配置管理：临时更改系统或应用配置，并在完成操作后恢复原始配置。

  7. 性能监控：使用上下文管理器来测量代码块的执行时间、内存使用等性能指标。

通过这些实际应用场景，上下文管理器可以大大提高代码的可读性、可维护性和健壮性。继续练习和探索，你会发现更多上下文管理器的妙用！

🌟 终极挑战：魔法学院管理系统 ⭐⭐⭐⭐⭐ ​

这是一个跨章节的综合项目，需要运用你学到的所有Python知识来构建一个完整的魔法学院管理系统。这个项目将测试你对Python编程的全面掌握程度。

项目要求 ​

创建一个magic_academy.py文件，实现以下功能：

1. 学生管理系统 ​

  • 使用类和继承创建学生类层次结构
  • 实现不同的魔法专业（变形术、魔药学、占卜学等）
  • 使用属性装饰器管理学生信息

2. 课程管理系统 ​

  • 使用装饰器实现权限检查和日志记录
  • 创建课程类，包含课程信息和学生名单
  • 实现选课和退课功能

3. 文件操作与数据持久化 ​

  • 使用上下文管理器安全地读写学生和课程数据
  • 实现数据导入导出功能（JSON格式）
  • 处理文件操作异常

4. 正则表达式应用 ​

  • 验证学生邮箱格式
  • 解析课程时间表
  • 搜索和过滤学生信息

5. 异常处理与日志 ​

  • 创建自定义异常类
  • 实现完整的错误处理机制
  • 使用上下文管理器管理日志文件

核心类设计 ​

python
class Student:
    """学生基类"""
    def __init__(self, name, student_id, email):
        self.name = name
        self.student_id = student_id
        self.email = email
        self.courses = []

    @property
    def email(self):
        return self._email

    @email.setter
    def email(self, value):
        # 使用正则表达式验证邮箱格式
        if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', value):
            raise ValueError("无效的邮箱格式")
        self._email = value

class MagicStudent(Student):
    """魔法学院学生"""
    def __init__(self, name, student_id, email, house, specialty):
        super().__init__(name, student_id, email)
        self.house = house  # 学院（格兰芬多、拉文克劳等）
        self.specialty = specialty  # 专业
        self.magic_level = 1

class Course:
    """课程类"""
    def __init__(self, course_id, name, teacher, max_students=30):
        self.course_id = course_id
        self.name = name
        self.teacher = teacher
        self.max_students = max_students
        self.students = []

    def enroll_student(self, student):
        if len(self.students) >= self.max_students:
            raise CourseFullError("课程已满")
        if student in self.students:
            raise AlreadyEnrolledError("学生已选修此课程")
        self.students.append(student)
        student.courses.append(self)

装饰器和上下文管理器实现 ​

python
import functools
import time
from datetime import datetime
import json
import re

def log_operation(func):
    """操作日志装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = func(*args, **kwargs)
            end_time = time.time()
            print(f"[{datetime.now()}] 操作 {func.__name__} 成功，耗时: {end_time - start_time:.2f}秒")
            return result
        except Exception as e:
            print(f"[{datetime.now()}] 操作 {func.__name__} 失败: {e}")
            raise
    return wrapper

class DataFileManager:
    """数据文件管理器上下文管理器"""
    def __init__(self, filename, mode='r'):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        try:
            self.file = open(self.filename, self.mode, encoding='utf-8')
            return self.file
        except FileNotFoundError:
            raise FileNotFoundError(f"文件 {self.filename} 不存在")
        except PermissionError:
            raise PermissionError(f"没有权限访问文件 {self.filename}")

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
            print(f"文件 {self.filename} 已安全关闭")
        return False  # 不抑制异常

挑战任务 ​

  1. 实现完整的学生和课程管理系统
  2. 添加数据验证和错误处理
  3. 实现文件操作的上下文管理器
  4. 创建合适的装饰器进行日志记录和权限检查
  5. 使用正则表达式验证输入数据
  6. 设计合理的类继承结构

这个综合项目将帮助你将所学的所有Python知识融会贯通，是检验学习成果的最佳方式！

© 2025 技术博客. All rights reserved by 老周有AI

Powered by VitePress