Python进阶之with语句优雅管理文件资源的完整指南

class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        print(f"✅ 正在打开文件: {self.filename}")
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
            print(f"✅ 已关闭文件: {self.filename}")
        # 可选：异常处理逻辑
        if exc_type:
            print(f"❌ 发生异常: {exc_val}")
        return False  # 不抑制异常

# 使用示例
with FileManager("example.txt", "w") as f:
    f.write("Hello, World!\n")
    f.write("This is a test.\n")
    # 模拟异常
    # raise ValueError("Something went wrong!")

# ❌ 坏习惯：手动管理
f = open("data.txt", "r")
content = f.read()
f.close()

# ✅ 好做法：使用 with
with open("data.txt", "r") as f:
    content = f.read()
    print(content)
# file 自动关闭！

with open("log.txt", "a") as log_file:
    log_file.write(f"[{datetime.now()}] User logged in\n")
    log_file.write(f"[{datetime.now()}] Data processed successfully\n")

import os

def process_large_file(filename):
    total_lines = 0
    total_size = 0

    with open(filename, "r", encoding="utf-8") as file:
        for line in file:
            total_lines += 1
            total_size += len(line)
            if total_lines % 1000 == 0:
                print(f"📦 已处理 {total_lines} 行，当前大小: {total_size} bytes")

    print(f"📊 总计: {total_lines} 行，总大小: {total_size} bytes")

# 调用函数
process_large_file("huge_log.txt")

import random

with open("test.txt", "w") as f:
    f.write("First line\n")
    # 随机抛出异常
    if random.choice([True, False]):
        raise RuntimeError("Simulated error during file write!")
    f.write("Second line\n")

from contextlib import contextmanager
import json
import pickle
from datetime import datetime

@contextmanager
def safe_file_handler(filename, mode="r", encoding="utf-8"):
    """
    安全的文件上下文管理器，支持自动重试和日志记录
    """
    file_obj = None
    retry_count = 0
    max_retries = 3

    while retry_count < max_retries:
        try:
            print(f"🔄 尝试打开文件: {filename} (第 {retry_count + 1} 次)")
            file_obj = open(filename, mode, encoding=encoding)
            break  # 成功打开，跳出循环
        except (IOError, OSError) as e:
            retry_count += 1
            print(f"⚠️ 打开失败: {e}, {max_retries - retry_count} 次重试机会")
            if retry_count >= max_retries:
                raise RuntimeError(f"❌ 无法打开文件 {filename}，已重试 {max_retries} 次") from e

    try:
        yield file_obj
    except Exception as e:
        print(f"💥 处理过程中发生异常: {e}")
        raise
    finally:
        if file_obj and not file_obj.closed:
            file_obj.close()
            print(f"✅ 已关闭文件: {filename}")

# ✅ 使用示例
if __name__ == "__main__":
    # 写入数据
    with safe_file_handler("config.json", "w") as f:
        data = {"app": "MyApp", "version": "1.0", "created_at": str(datetime.now())}
        json.dump(data, f, indent=2)

    # 读取数据
    with safe_file_handler("config.json", "r") as f:
        config = json.load(f)
        print("📄 配置加载成功:", config)

# 错误认知：only for files
with open("file.txt", "r") as f:
    pass

# 数据库连接
import sqlite3

with sqlite3.connect("example.db") as conn:
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER, name TEXT)")
    cursor.execute("INSERT INTO users VALUES (?, ?)", (1, "Alice"))
    conn.commit()

# 网络请求（模拟）
import requests

with requests.Session() as session:
    resp = session.get("https://httpbin.org/get")
    print(resp.json())

# ❌ 错误写法（非必须）
with open("in.txt", "r") as fin:
    with open("out.txt", "w") as fout:
        for line in fin:
            fout.write(line.upper())

# ✅ 推荐写法：链式嵌套
with open("in.txt", "r") as fin, open("out.txt", "w") as fout:
    for line in fin:
        fout.write(line.upper())

from contextlib import contextmanager
import time

@contextmanager
def timer():
    start = time.time()
    print("⏱️ 开始计时...")
    try:
        yield
    finally:
        end = time.time()
        print(f"✅ 任务耗时: {end - start:.4f} 秒")

# 使用示例
with timer():
    time.sleep(1)
    print("⏳ 任务完成")

import time
import os

def test_manual_close():
    files = []
    start = time.time()
    for i in range(1000):
        f = open(f"data_{i}.txt", "w")
        f.write(f"Line {i}\n")
        files.append(f)
    for f in files:
        f.close()
    print(f"📝 手动关闭耗时: {time.time() - start:.4f} 秒")

def test_with_statement():
    start = time.time()
    for i in range(1000):
        with open(f"data_{i}.txt", "w") as f:
            f.write(f"Line {i}\n")
    print(f"✨ with 语句耗时: {time.time() - start:.4f} 秒")

# 执行测试（注意：首次运行会创建文件）
test_manual_close()
test_with_statement()