为什么需要上下文管理器？

想象这样一个场景：你正在编写一个处理文件的程序。传统的写法是这样的：

1

2

3

4

f = open('data.txt', 'r')

data = f.read()

# 如果这里发生异常，文件就不会被关闭

f.close()

这种写法存在严重问题：如果在读取文件时抛出异常，f.close()可能永远不会被执行，导致资源泄漏。即使使用try-finally块，代码也显得冗长：

1

2

3

4

5

f = open('data.txt', 'r')

try:

    data = f.read()

finally:

    f.close()  # 确保无论如何都会关闭

Python的with语句让这一切变得简单优雅：

1

2

3

with open('data.txt', 'r') as f:

    data = f.read()

# 文件在这里自动关闭，即使有异常发生

二、上下文管理器的工作原理

上下文管理器的核心是两个特殊方法：__enter__和__exit__。

1. __enter__方法

当程序执行流进入with语句块时，会调用__enter__方法。它返回的值会被赋给as后面的变量。

2. __exit__方法

当程序离开with语句块时（无论是正常结束还是异常退出），都会调用__exit__方法。它接收三个参数：

  • exc_type：异常类型
  • exc_val：异常值
  • exc_tb：异常追踪信息

如果没有异常发生，这三个参数都是None。

三、自定义上下文管理器

让我们通过实现一个自定义的数据库连接管理器来深入理解：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

import sqlite3

from typing import Optional

class DatabaseConnection:

    """数据库连接上下文管理器"""

     

    def __init__(self, db_path: str):

        self.db_path = db_path

        self.connection: Optional[sqlite3.Connection] = None

     

    def __enter__(self) -> sqlite3.Connection:

        """建立连接并返回"""

        print(f"🔗 正在连接数据库: {self.db_path}")

        self.connection = sqlite3.connect(self.db_path)

        return self.connection

     

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:

        """关闭连接，处理异常"""

        if self.connection:

            if exc_type is None:

                # 没有异常，提交事务

                print("✅ 无异常，提交事务")

                self.connection.commit()

            else:

                # 发生异常，回滚事务

                print(f"❌ 发生异常: {exc_val}")

                self.connection.rollback()

             

            print("🔒 关闭数据库连接")

            self.connection.close()

         

        # 返回False表示不抑制异常，继续抛出

        return False

# 使用示例

with DatabaseConnection('example.db') as conn:

    cursor = conn.cursor()

    cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")

    cursor.execute("INSERT INTO users (name) VALUES ('Alice')")

四、使用contextlib简化实现

对于简单的场景，Python的contextlib模块提供了更简洁的方式。

1. @contextmanager装饰器

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

from contextlib import contextmanager

from typing import Generator

import time

@contextmanager

def timer(name: str) -> Generator[None, None, None]:

    """计时器上下文管理器"""

    start = time.time()

    print(f"⏱️ 开始计时: {name}")

    try:

        yield  # 这里会执行with语句块内的代码

    finally:

        elapsed = time.time() - start

        print(f"⏹️ 结束计时: {name}, 耗时 {elapsed:.4f} 秒")

# 使用示例

with timer("数据处理"):

    time.sleep(1)

    result = sum(range(1000000))

    print(f"计算结果: {result}")

2. 嵌套上下文管理器

1

2

3

4

5

6

7

8

9

from contextlib import ExitStack

# 同时管理多个资源

with ExitStack() as stack:

    file1 = stack.enter_context(open('file1.txt', 'w'))

    file2 = stack.enter_context(open('file2.txt', 'w'))

     

    file1.write("Hello")

    file2.write("World")

五、实战案例

案例1：重定向标准输出

1

2

3

4

5

6

7

8

9

from contextlib import redirect_stdout

import io

output = io.StringIO()

with redirect_stdout(output):

    print("这条消息不会显示在控制台")

     

captured = output.getvalue()

print(f"捕获的内容: {captured}")

案例2：临时修改环境变量

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

import os

from contextlib import contextmanager

@contextmanager

def temp_env_var(key: str, value: str):

    old_value = os.environ.get(key)

    os.environ[key] = value

    try:

        yield

    finally:

        if old_value is None:

            del os.environ[key]

        else:

            os.environ[key] = old_value

with temp_env_var('MY_VAR', 'temp_value'):

    print(os.environ.get('MY_VAR'))

案例3：线程锁的优雅使用

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

import threading

from concurrent.futures import ThreadPoolExecutor

class SafeCounter:

    def __init__(self):

        self.value = 0

        self._lock = threading.Lock()

     

    def increment(self):

        with self._lock:  # 自动获取和释放锁

            self.value += 1

counter = SafeCounter()

with ThreadPoolExecutor(max_workers=10) as executor:

    futures = [executor.submit(counter.increment) for _ in range(1000)]

     

print(f"最终计数: {counter.value}")  # 输出: 1000

六、suppress：优雅地忽略异常

1

2

3

4

5

6

7

8

9

10

11

from contextlib import suppress

# 传统写法

try:

    os.remove('可能不存在的文件.txt')

except FileNotFoundError:

    pass

# 使用suppress更简洁

with suppress(FileNotFoundError):

    os.remove('可能不存在的文件.txt')

七、最佳实践与注意事项

  1. 始终确保资源释放：在__exit__或finally块中清理资源
  2. 正确处理异常：根据需要决定是否抑制异常
  3. 使用@contextmanager：对于简单场景，它比类更简洁
  4. 善用标准库：contextlib提供了很多实用工具
  5. 文档说明：上下文管理器的行为应当在文档字符串中清晰说明

总结

# python # 数据库 # oracle
Logo
亚马逊云科技技术品牌专区

更多推荐

2026年量化工具选择，要跟着能力基础走

读者应理解，选择工具前要先判断自己的能力基础和当前任务。工具应该帮助自己补上当前最关键的缺口，而不是替代学习顺序本身。

avatar 亚马逊云科技技术品牌专区

加州理工 CS367 C 语言系统编程笔记（一）

C语言是一种强大且广泛使用的编程语言，尤其在系统编程领域。本节课我们将学习C语言的基础语法，包括如何编写“Hello, World!”程序、声明变量以及使用基本数据类型。我们将通过实际的代码示例来加深理解。本节课中我们一起学习了C语言的基础语法，包括如何编写和运行一个简单的C程序、声明变量以及使用基本数据类型。我们还介绍了如何使用Shell环境来编译和运行程序。通过对比Java，你可以看到C语言在

avatar 亚马逊云科技技术品牌专区

近期零基础学量化，先按基础差异拆学习顺序

读者应明白，拆解学习顺序时要先看自己的基础缺口，再决定哪一段需要放慢。不同流程阶段都有不同检查任务，学习路线要能反映这些差异。

avatar 亚马逊云科技技术品牌专区