Python with 关键字

file = open("data.txt", "w")
try:
    file.write("Hello, World!")
    # 如果这里发生异常，或者开发者忘记写 finally，文件将不会被关闭
    # 导致文件描述符泄漏 (File Descriptor Leak)
finally:
    file.close()  # 必须手动确保关闭

with open("data.txt", "w") as file:
    file.write("Hello, World!")
# 离开 with 代码块时，文件会自动、安全地关闭，无论是否发生异常

# 当你写下这段代码时：
with EXPR as VAR:
    BLOCK

# Python 解释器在底层实际执行的是：
manager = EXPR
enter = type(manager).__enter__
exit = type(manager).__exit__
value = enter(manager)
hit_except = False

try:
    VAR = value
    BLOCK  # 执行你的代码块
except:
    hit_except = True
    # 如果 __exit__ 返回 False，异常会继续向外抛出
    if not exit(manager, *sys.exc_info()):
        raise
finally:
    # 如果没有发生异常，也会调用 __exit__，且三个参数均为 None
    if not hit_except:
        exit(manager, None, None, None)

import time

class Timer:
    def __init__(self, name="Operation"):
        self.name = name

    def __enter__(self):
        self.start_time = time.perf_counter()
        print(f"[{self.name}] 开始执行...")
        return self  # 返回 self，以便在 with 块中使用

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.end_time = time.perf_counter()
        elapsed = self.end_time - self.start_time
        print(f"[{self.name}] 执行完毕，耗时: {elapsed:.4f} 秒")

        # 如果发生异常，可以在这里记录日志
        if exc_type is not None:
            print(f"[{self.name}] 发生异常: {exc_val}")

        # 返回 False，让异常正常向外抛出
        return False

# 使用自定义的上下文管理器
with Timer("数据处理") as timer:
    time.sleep(1)
    print("正在处理数据...")
    # 如果取消下一行的注释，异常会被 __exit__ 捕获并打印，然后继续向外抛出
    # raise ValueError("模拟错误")

from contextlib import contextmanager

@contextmanager
def managed_file(filename, mode):
    print(f"正在打开文件: {filename}")
    f = open(filename, mode)
    try:
        # yield 的值将赋给 as 后面的变量
        yield f
    finally:
        print(f"正在关闭文件: {filename}")
        f.close()

# 使用方式与内置的 open 完全一致
with managed_file("test.txt", "w") as f:
    f.write("Hello from contextmanager!")

from contextlib import closing
import urllib.request

# urllib.request.urlopen 返回的对象有 close() 方法，但在某些旧版本中不是完美的上下文管理器
with closing(urllib.request.urlopen('https://www.python.org')) as page:
    for line in page:
        print(line.decode('utf-8'))
# 离开 with 块时，closing 会自动调用 page.close()

import os

filename = "temp_file.txt"

# ❌ 传统写法：空的 except 块容易被误认为是遗漏了处理逻辑
try:
    os.remove(filename)
except FileNotFoundError:
    pass

# ✅ 推荐写法：清晰表达“我就是要忽略这个特定的异常”
from contextlib import suppress
with suppress(FileNotFoundError):
    os.remove(filename)

import io
from contextlib import redirect_stdout

# 捕获 print 的输出
f = io.StringIO()
with redirect_stdout(f):
    print("这段文字不会打印到控制台，而是进入了 StringIO")

output = f.getvalue()
print(f"捕获到的内容: {repr(output)}")  # '这段文字不会打印到控制台，而是进入了 StringIO\n'

from contextlib import ExitStack

filenames = ["file1.txt", "file2.txt", "file3.txt"]

# ❌ 传统写法：如果 filenames 有 100 个，代码将无法编写
# with open(filenames[0]) as f0, open(filenames[1]) as f1 ...

# ✅ 推荐写法：使用 ExitStack
with ExitStack() as stack:
    # 动态打开所有文件，并将文件对象保存在列表中
    files = [stack.enter_context(open(fname, 'w')) for fname in filenames]

    # 现在可以安全地操作所有文件
    for f in files:
        f.write("Data\n")

# 离开 with 块时，ExitStack 会自动、安全地按逆序关闭所有打开的文件。
# 即使在打开 file2.txt 时发生异常，file1.txt 也会被正确关闭。

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

with (
    open("file1.txt", "r") as f1,
    open("file2.txt", "r") as f2,
    Timer("Process") as timer
):
    # 执行操作
    pass

   import sqlite3
   conn = sqlite3.connect("test.db")
   with conn:  # 这里的 with 管理的是事务，而不是连接本身
       conn.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
   # 离开块时自动 commit，如果发生异常则自动 rollback

with open("test.txt", "w") as f:
    f.write("data")
    my_var = "I am inside with"

print(my_var)  # 输出: "I am inside with" (变量并未被销毁)

class BadManager:
    def __enter__(self): return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Cleaning up...")
        return True  # ⚠️ 危险！这会吞掉 with 块内发生的所有异常！

with BadManager():
    raise ValueError("这个错误将被静默吞掉，极难调试！")

# ❌ 反模式：print 不需要清理
# with print("Hello"): pass 

# ✅ 正确：直接调用
print("Hello")

import requests

with requests.Session() as session:
    # Session 会在退出时自动关闭底层的连接池
    response = session.get("https://api.github.com")