Python with语句详解

1. with语句的基本原理

1.1 执行流程

1. 执行上下文表达式：计算 with后面的表达式，获取上下文管理器对象

2. 调用 __enter__()：进入上下文时调用，返回值赋给 as后面的变量

3. 执行代码块：执行 with语句体内的代码

4. 调用 __exit__()：无论代码块是否发生异常，都会调用此方法进行清理

    

1.2 异常处理机制

• exc_type：异常类型（无异常时为 None）

• exc_value：异常值

• exc_traceback：异常回溯信息

   

2. with语句的基本用法

2.1 基本语法

with context_expression [as target]:
    # with-body
    statements

2.2 与传统写法的对比

file = open("data.txt", "r")
try:
    data = file.read()
finally:
    file.close()  # 必须手动关闭

with open("data.txt", "r") as file:
    data = file.read()  # 文件自动关闭

3. with语句的常见应用场景

3.1 文件操作

# 读取文件
with open("example.txt", "r") as file:
    content = file.read()

# 写入文件
with open("output.txt", "w") as file:
    file.write("Hello, World!")

# 处理大文件（逐行读取）
with open("large_file.txt", "r") as file:
    for line in file:
        process_line(line)

3.2 数据库连接管理

import sqlite3

with sqlite3.connect("database.db") as conn:
    cursor = conn.cursor()
    cursor****cute("SELECT * FROM users")
    results = cursor.fetchall()
# 连接自动关闭[6](@ref)

3.3 线程锁管理

import threading

lock = threading.Lock()
shared_data = 0

def increment():
    global shared_data
    with lock:  # 自动获取和释放锁
        shared_data += 1

3.4 临时文件和目录

import tempfile

# 临时文件（自动删除）
with tempfile.NamedTemporaryFile(delete=True) as temp_file:
    temp_file.write(b"Some data")
    temp_file.seek(0)
    print(temp_file.read())

# 临时目录（自动清理）
with tempfile.TemporaryDirectory() as temp_dir:
    # 使用临时目录
    print(f"Working in {temp_dir}")

4. 自定义上下文管理器

4.1 基于类的实现

class DatabaseConnection:
    def __init__(self, db_name):
        self.db_name = db_name
        self.connection = None
    
    def __enter__(self):
        print(f"连接数据库: {self.db_name}")
        self.connection = f"连接到{self.db_name}"  # 模拟连接
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"关闭数据库连接: {self.db_name}")
        self.connection = None
        # 返回False，让异常传播
        return False

# 使用自定义上下文管理器
with DatabaseConnection("my_db") as db:
    print(f"使用: {db}")

4.2 使用 contextlib模块

from contextlib import contextmanager
import time

@contextmanager
def timer(name):
    start = time.time()
    try:
        yield  # 执行with块内的代码
    finally:
        end = time.time()
        print(f"{name} 耗时: {end-start:.2f}秒")

# 使用
with timer("数据处理"):
    time.sleep(1.5)

4.3 高级异常处理

class IgnoreException:
    def __init__(self, exc_type):
        self.exc_type = exc_type
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type == self.exc_type:
            print(f"忽略异常: {exc_type.__name__}")
            return True  # 抑制异常
        return False  # 其他异常正常传播

# 忽略特定异常
with IgnoreException(ValueError):
    raise ValueError("这个异常会被忽略")

5. 高级用法和技巧

5.1 多个上下文管理器

# 同时打开两个文件
with open("input.txt", "r") as infile, open("output.txt", "w") as outfile:
    content = infile.read()
    outfile.write(content.upper())

# 多行写法（更清晰）
with open("file1.txt", "r") as f1, \
     open("file2.txt", "r") as f2, \
     open("result.txt", "w") as f3:
    # 处理多个文件
    pass

5.2 嵌套上下文管理器

with open("source.txt", "r") as src:
    with open("destination.txt", "w") as dst:
        for line in src:
            dst.write(line.upper())

5.3 测量执行时间

from contextlib import contextmanager

@contextmanager
def profile(operation_name):
    start = time.perf_counter()
    try:
        yield
    finally:
        end = time.perf_counter()
        print(f"{operation_name} 耗时: {end-start:.4f}秒")

with profile("复杂计算"):
    # 需要测量时间的代码
    time.sleep(0.5)

6. with语句在 Django 框架中的优秀实践

6.1 数据库事务管理

from django.db import transaction

def update_user_profile(user_id, new_data):
    with transaction.atomic():  # 创建事务
        user = User.objects.select_for_update().get(id=user_id)
        user.name = new_data['name']
        user.email = new_data['email']
        user.save()
        
        # 相关操作也在同一事务中
        Profile.objects.filter(user=user).update(**new_data['profile'])
    # 事务自动提交（或无异常时回滚）

6.2 缓存管理

from django.core.cache import cache

def get_expensive_data(key):
    # 尝试从缓存获取，避免重复计算
    with cache.lock(f'lock-{key}', timeout=10):  # 缓存锁
        data = cache.get(key)
        if data is None:
            data = expensive_calculation()
            cache.set(key, data, timeout=3600)
        return data

6.3 请求超时处理

import signal
from contextlib import contextmanager

@contextmanager
def timeout(seconds):
    def timeout_handler(signum, frame):
        raise TimeoutError("操作超时")
    
    old_handler = signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

# 在Django视图中使用
def slow_operation_view(request):
    try:
        with timeout(5):  # 5秒超时
            result = slow_database_query()
            return JsonResp***e({'result': result})
    except TimeoutError:
        return JsonResp***e({'error': '请求超时'}, status=408)

6.4 临时切换设置

from django.test import override_settings
from contextlib import contextmanager

@contextmanager
def temporary_language(language_code):
    with override_settings(LANGUAGE_CODE=language_code):
        from django.utils import translation
        translation.activate(language_code)
        try:
            yield
        finally:
            translation.deactivate()

# 临时切换语言
def multi_language_report(user):
    with temporary_language('en'):
        english_report = generate_report(user)
    
    with temporary_language('zh-hans'):
        chinese_report = generate_report(user)
    
    return english_report, chinese_report

6.5 数据库连接管理

from django.db import connection

class QueryLogger:
    def __enter__(self):
        self.queries_before = len(connection.queries)
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.queries_after = len(connection.queries)
        new_queries = connection.queries[self.queries_before:self.queries_after]
        print(f"执行了 {len(new_queries)} 条**L查询")
        for query in new_queries:
            print(f"**L: {query['sql']}, 耗时: {query['time']}s")

# 在视图中分析查询性能
def optimized_view(request):
    with QueryLogger():
        users = User.objects.select_related('profile').filter(is_active=True)
        data = list(users)  # 强制查询执行
    return JsonResp***e({'users': len(data)})

7. 注意事项和最佳实践

7.1 性能考虑

• with语句有轻微的性能开销，但在大多数场景下可忽略

• 在性能敏感的循环中，考虑在循环外部使用 with

7.2 异常处理最佳实践

• 在 __exit__中谨慎返回 True（抑制异常），除非有明确理由

• 确保在 __exit__中不抛出新的异常

7.3 资源泄漏预防

• 如果 __enter__中获取资源失败，__exit__不会被调用

• 对关键资源应有备用清理机制

总结