> ## Documentation Index
> Fetch the complete documentation index at: https://docs.akria.net/llms.txt
> Use this file to discover all available pages before exploring further.

# Day 09 | 数据结构与装饰器

> 字典高级操作、异常处理与装饰器在运维中的应用

<Info>
  **今日目标**：深入理解 Python 的核心数据结构 —— **字典 (Dictionary)**，并学会使用 **装饰器 (Decorator)** 来优雅地增加重试机制和鉴权逻辑。今天不只是写代码，而是要彻底理解 **"字典为什么是 Python 的灵魂"**、**"装饰器如何优雅地扩展函数功能"** 以及 **"异常处理如何让代码更健壮"**。
</Info>

## <span style={{ display: 'inline-flex', alignItems: 'center', gap: '8px' }}><Icon icon="book" size={26} color="#059669" /> 学习内容 (30 mins)</span>

在开始写代码前，先搞懂这些核心概念，否则后面的代码你会看得云里雾里。

<AccordionGroup>
  <Accordion title="字典进阶 (15 mins)" icon="database">
    **什么是字典？**

    字典（Dictionary）是 Python 中的键值对（key-value）数据结构，类似于其他语言中的 Map 或 Hash Table。

    **为什么字典是 Python 的灵魂？**

    * **查找速度快**：O(1) 时间复杂度，比列表的 O(n) 快得多
    * **灵活性强**：可以存储任意类型的数据
    * **应用广泛**：JSON、配置文件、API 响应都是字典结构
    * **Pythonic**：字典推导式让代码更简洁优雅

    **字典核心操作详解**：

    * **`get(key, default)`**：
      * 作用：安全获取字典值，如果 key 不存在返回默认值
      * 为什么需要：`dict['key']` 如果 key 不存在会报 `KeyError`，`get()` 更安全
      * 示例：`data.get('port', 3306)` 如果 'port' 不存在，返回 3306
    * **`setdefault(key, default)`**：
      * 作用：如果 key 不存在，设置默认值并返回；如果存在，返回原值
      * 应用场景：初始化嵌套字典结构
      * 示例：`data.setdefault('config', {})` 确保 'config' 键存在
    * **`update(other_dict)`**：
      * 作用：用另一个字典更新当前字典
      * 应用场景：合并配置、更新参数
      * 注意：相同的 key 会被覆盖
    * **字典推导式**：
      * 语法：`{key_expr: value_expr for item in iterable if condition}`
      * 优势：一行代码完成筛选和转换，比 for 循环更 Pythonic
      * 示例：`{k: v*2 for k, v in data.items() if v > 10}`
  </Accordion>

  <Accordion title="装饰器 (10 mins)" icon="magic">
    **什么是装饰器？**

    装饰器（Decorator）是一个函数，它接受一个函数作为参数，并返回一个新的函数，用于在不修改原函数代码的情况下扩展功能。

    **为什么需要装饰器？**

    * **代码复用**：将通用功能（如重试、日志、计时）抽取出来
    * **关注点分离**：业务逻辑和横切关注点（cross-cutting concerns）分离
    * **优雅扩展**：不需要修改原函数，只需添加装饰器
    * **Pythonic**：装饰器是 Python 的特色功能，让代码更优雅

    **装饰器原理**：

    * **闭包（Closure）**：内部函数可以访问外部函数的变量
    * **函数是一等公民**：函数可以作为参数传递，也可以作为返回值
    * **语法糖**：`@decorator` 等价于 `func = decorator(func)`

    **装饰器应用场景**：

    * **重试机制**：网络请求失败自动重试
    * **性能监控**：统计函数执行时间
    * **权限校验**：检查用户是否有权限执行函数
    * **日志记录**：自动记录函数调用和参数
  </Accordion>

  <Accordion title="异常处理 (5 mins)" icon="shield">
    **什么是异常处理？**

    异常处理是捕获和处理程序运行时的错误，防止程序崩溃。

    **为什么必须正确处理异常？**

    * **程序健壮性**：即使出错也能优雅处理，不会崩溃
    * **资源清理**：确保文件、数据库连接等资源被正确释放
    * **用户体验**：给用户友好的错误提示，而不是程序崩溃

    **异常处理规范**：

    * **永远不要写裸露的 `except:`**：
      * 会捕获所有异常，包括系统退出（SystemExit）、键盘中断（KeyboardInterrupt）等
      * 至少捕获 `Exception`：`except Exception as e:`
    * **使用 `finally` 清理资源**：
      * 无论是否发生异常，`finally` 块都会执行
      * 适合关闭文件、数据库连接等操作
    * **具体异常优于通用异常**：
      * `except KeyError:` 比 `except Exception:` 更精确
      * 可以针对不同异常做不同处理
  </Accordion>
</AccordionGroup>

***

## <span style={{ display: 'inline-flex', alignItems: 'center', gap: '8px' }}><Icon icon="code" size={26} color="#059669" /> 代码任务 (90 mins)</span>

<Steps>
  <Step title="环境准备">
    确保虚拟环境已激活，并安装必要的包：

    ```bash theme={null}
    # 确保虚拟环境已激活（提示符前有 (.venv)）
    # 如果未激活，执行：
    source .venv/bin/activate

    # 今天不需要安装额外包，使用 Python 标准库即可
    ```
  </Step>

  <Step title="任务 A：字典处理实战">
    编写 `09_dict_ops.py`，处理模拟的服务器监控数据。

    **任务分解**：

    1. 使用字典安全访问方法处理监控数据
    2. 使用字典推导式筛选高负载服务器
    3. 处理嵌套字典结构

    <CodeGroup>
      ```python 09_dict_ops.py theme={null}
      #!/usr/bin/env python3
      """
      Day 09 - 字典操作实战
      演示字典的安全访问、推导式和嵌套结构处理
      """

      # ========== 1. 基础字典操作 ==========
      print("=== 基础字典操作 ===")

      # 模拟 Prometheus 返回的监控数据
      # 每个字典代表一台服务器的监控指标
      metrics = [
          {"host": "web-01", "cpu": 45, "mem": 60},
          {"host": "db-01",  "cpu": 85, "mem": 90},  # Alert! 高负载
          {"host": "web-02", "cpu": 10, "mem": 20},
          {"host": "cache-01", "cpu": 30, "mem": 40},
      ]

      # 1.1 提取所有主机名（列表推导式）
      # 从每个字典中提取 'host' 键的值
      hosts = [m["host"] for m in metrics]
      print(f"监控的主机列表: {hosts}")
      # 输出: ['web-01', 'db-01', 'web-02', 'cache-01']

      # 1.2 筛选高负载节点（字典推导式）
      # 条件：CPU > 80 或 内存 > 80
      # 结果：{主机名: CPU使用率}
      alerts = {
          m["host"]: m["cpu"] 
          for m in metrics 
          if m["cpu"] > 80 or m["mem"] > 80
      }
      print(f"高负载服务器: {alerts}")
      # 输出: {'db-01': 85}

      # ========== 2. 字典安全访问 ==========
      print("\n=== 字典安全访问 ===")

      # 模拟 API 返回的配置数据（可能缺少某些字段）
      api_response = {
          "host": "web-01",
          "port": 8080,
          # 注意：没有 'timeout' 和 'retries' 字段
      }

      # 2.1 错误方式：直接访问可能不存在的键
      # port = api_response['port']  # 如果 'port' 不存在，会报 KeyError

      # 2.2 正确方式：使用 get() 方法
      port = api_response.get('port', 80)  # 如果不存在，返回默认值 80
      timeout = api_response.get('timeout', 30)  # 默认超时 30 秒
      retries = api_response.get('retries', 3)  # 默认重试 3 次

      print(f"端口: {port}, 超时: {timeout}秒, 重试: {retries}次")
      # 输出: 端口: 8080, 超时: 30秒, 重试: 3次

      # ========== 3. 嵌套字典处理 ==========
      print("\n=== 嵌套字典处理 ===")

      # 模拟复杂的配置结构
      config = {
          "database": {
              "host": "localhost",
              "port": 3306,
              "name": "infra_db"
          },
          "redis": {
              "host": "localhost",
              "port": 6379
          }
      }

      # 3.1 安全访问嵌套字典
      # 错误方式：config['database']['port']  # 如果 'database' 不存在，会报 KeyError

      # 正确方式：使用 get() 链式调用
      db_port = config.get('database', {}).get('port', 3306)
      # 解释：
      # - config.get('database', {}) 如果 'database' 不存在，返回空字典 {}
      # - .get('port', 3306) 从返回的字典中获取 'port'，不存在则返回 3306

      print(f"数据库端口: {db_port}")

      # 3.2 使用 setdefault() 初始化嵌套结构
      # 场景：动态构建配置字典
      server_config = {}

      # 如果 'web' 键不存在，创建一个空字典
      server_config.setdefault('web', {})['host'] = 'web-01'
      server_config.setdefault('web', {})['port'] = 8080

      # 如果 'db' 键不存在，创建一个空字典
      server_config.setdefault('db', {})['host'] = 'db-01'
      server_config.setdefault('db', {})['port'] = 3306

      print(f"服务器配置: {server_config}")
      # 输出: {'web': {'host': 'web-01', 'port': 8080}, 'db': {'host': 'db-01', 'port': 3306}}

      # ========== 4. 字典更新与合并 ==========
      print("\n=== 字典更新与合并 ===")

      # 默认配置
      default_config = {
          "timeout": 30,
          "retries": 3,
          "debug": False
      }

      # 用户自定义配置（可能只覆盖部分字段）
      user_config = {
          "timeout": 60,  # 覆盖默认值
          "debug": True   # 覆盖默认值
          # retries 使用默认值
      }

      # 合并配置：用户配置优先
      final_config = default_config.copy()  # 复制默认配置，避免修改原字典
      final_config.update(user_config)  # 用用户配置更新

      print(f"最终配置: {final_config}")
      # 输出: {'timeout': 60, 'retries': 3, 'debug': True}
      ```
    </CodeGroup>

    **代码解释**：

    * **字典推导式**：`{k: v for item in list if condition}` 一行完成筛选和转换
    * **`get()` 方法**：安全访问字典，避免 `KeyError`
    * **`setdefault()` 方法**：初始化嵌套字典结构
    * **`update()` 方法**：合并字典，相同 key 会被覆盖

    **运行脚本**：

    ```bash theme={null}
    python 09_dict_ops.py
    ```

    **验证步骤**：

    1. 脚本能正常运行，无语法错误
    2. 输出包含所有主机列表
    3. 输出包含高负载服务器（db-01）
    4. 输出包含安全访问的配置信息
    5. 输出包含嵌套字典处理结果

    **常见错误**：

    <Warning>
      * ❌ `KeyError: 'port'` - 直接访问不存在的键，应使用 `get()` 方法
      * ❌ `TypeError: 'NoneType' object is not subscriptable` - `get()` 返回 None 后继续访问，应提供默认值
      * ❌ `AttributeError: 'dict' object has no attribute 'append'` - 字典没有 `append()` 方法，应使用 `update()` 或直接赋值
    </Warning>
  </Step>

  <Step title="任务 B：编写重试装饰器">
    在运维脚本中，网络波动很常见。编写一个 `@retry` 装饰器，实现"报错自动重试 3 次"。

    **任务分解**：

    1. 理解装饰器的基本结构
    2. 实现重试逻辑
    3. 添加异常处理和日志

    <CodeGroup>
      ```python 09_decorator.py theme={null}
      #!/usr/bin/env python3
      """
      Day 09 - 装饰器实战
      实现一个通用的重试装饰器
      """

      import time
      import random
      from functools import wraps

      # ========== 装饰器实现 ==========
      def retry(max_retries=3, delay=1):
          """
          重试装饰器
          
          参数:
              max_retries: 最大重试次数（默认 3 次）
              delay: 每次重试之间的延迟（秒，默认 1 秒）
          
          使用示例:
              @retry(max_retries=3, delay=1)
              def connect_db():
                  # 连接数据库的逻辑
                  pass
          """
          def decorator(func):
              # 使用 @wraps 保留原函数的元信息（函数名、文档字符串等）
              @wraps(func)
              def wrapper(*args, **kwargs):
                  """
                  包装函数：实现重试逻辑
                  *args: 位置参数
                  **kwargs: 关键字参数
                  """
                  last_exception = None
                  
                  # 尝试执行函数，最多重试 max_retries 次
                  for attempt in range(max_retries):
                      try:
                          # 执行原函数
                          result = func(*args, **kwargs)
                          # 如果成功，直接返回结果
                          if attempt > 0:
                              print(f"✅ 第 {attempt + 1} 次尝试成功！")
                          return result
                      except Exception as e:
                          # 捕获异常，记录最后一次异常
                          last_exception = e
                          print(f"⚠️  第 {attempt + 1} 次尝试失败: {e}")
                          
                          # 如果不是最后一次尝试，等待后重试
                          if attempt < max_retries - 1:
                              print(f"⏳ 等待 {delay} 秒后重试...")
                              time.sleep(delay)
                  
                  # 所有重试都失败，抛出最后一次异常
                  print(f"❌ 所有 {max_retries} 次尝试都失败了")
                  raise last_exception
              
              return wrapper
          return decorator

      # ========== 使用示例 1：模拟不稳定的数据库连接 ==========
      @retry(max_retries=3, delay=1)
      def connect_database():
          """
          模拟数据库连接（70% 概率失败）
          """
          # 随机生成 0-1 之间的数
          # 如果小于 0.7（70% 概率），抛出异常
          if random.random() < 0.7:
              raise ConnectionError("数据库连接超时")
          
          # 30% 概率成功
          print("✅ 数据库连接成功！")
          return {"status": "connected", "host": "db-01"}

      # ========== 使用示例 2：模拟 API 请求 ==========
      @retry(max_retries=5, delay=2)
      def fetch_api_data(url: str) -> dict:
          """
          模拟 API 请求（60% 概率失败）
          """
          if random.random() < 0.6:
              raise TimeoutError(f"请求 {url} 超时")
          
          print(f"✅ 成功获取 {url} 的数据")
          return {"data": "success", "url": url}

      # ========== 使用示例 3：带参数的函数 ==========
      @retry(max_retries=3, delay=1)
      def process_server(server_name: str, port: int) -> str:
          """
          处理服务器配置（50% 概率失败）
          """
          if random.random() < 0.5:
              raise ValueError(f"服务器 {server_name}:{port} 配置错误")
          
          return f"服务器 {server_name}:{port} 配置成功"

      # ========== 主程序：测试装饰器 ==========
      if __name__ == "__main__":
          print("=" * 50)
          print("测试 1: 数据库连接（重试 3 次）")
          print("=" * 50)
          try:
              result = connect_database()
              print(f"结果: {result}")
          except Exception as e:
              print(f"最终失败: {e}")
          
          print("\n" + "=" * 50)
          print("测试 2: API 请求（重试 5 次）")
          print("=" * 50)
          try:
              result = fetch_api_data("https://api.example.com/data")
              print(f"结果: {result}")
          except Exception as e:
              print(f"最终失败: {e}")
          
          print("\n" + "=" * 50)
          print("测试 3: 服务器处理（带参数）")
          print("=" * 50)
          try:
              result = process_server("web-01", 8080)
              print(f"结果: {result}")
          except Exception as e:
              print(f"最终失败: {e}")
      ```
    </CodeGroup>

    **装饰器原理解析**：

    * **三层嵌套函数**：
      1. `retry(max_retries, delay)`：外层函数，接收装饰器参数
      2. `decorator(func)`：中层函数，接收被装饰的函数
      3. `wrapper(*args, **kwargs)`：内层函数，实际执行逻辑
    * **`@wraps(func)` 的作用**：
      * 保留原函数的元信息（函数名、文档字符串等）
      * 不使用 `@wraps` 的话，`func.__name__` 会变成 `wrapper`
    * **`*args` 和 `**kwargs`**：
      * `*args`：接收任意数量的位置参数
      * `**kwargs`：接收任意数量的关键字参数
      * 这样装饰器可以装饰任何签名的函数

    **运行脚本**：

    ```bash theme={null}
    python 09_decorator.py
    ```

    **验证步骤**：

    1. 脚本能正常运行，无语法错误
    2. 观察重试过程：失败后会等待并重试
    3. 如果最终成功，会显示"第 X 次尝试成功"
    4. 如果所有重试都失败，会抛出最后一次异常

    **常见错误**：

    <Warning>
      * ❌ `TypeError: retry() missing 1 required positional argument: 'func'` - 装饰器参数使用错误，应使用 `@retry(max_retries=3)`
      * ❌ `NameError: name 'wraps' is not defined` - 忘记导入 `from functools import wraps`
      * ❌ `AttributeError: 'function' object has no attribute '__name__'` - 可能是装饰器结构错误
    </Warning>
  </Step>

  <Step title="任务 C：异常处理实战">
    编写 `09_exception.py`，演示正确的异常处理方式。

    <CodeGroup>
      ```python 09_exception.py theme={null}
      #!/usr/bin/env python3
      """
      Day 09 - 异常处理实战
      演示正确的异常处理方式
      """

      # ========== 1. 基本异常处理 ==========
      print("=== 基本异常处理 ===")

      def safe_divide(a: float, b: float) -> float:
          """
          安全除法：处理除零错误
          """
          try:
              result = a / b
              return result
          except ZeroDivisionError:
              # 捕获具体的异常类型
              print("❌ 错误：除数不能为零")
              return 0.0
          except TypeError:
              # 可以捕获多种异常
              print("❌ 错误：参数类型不正确")
              return 0.0

      print(f"10 / 2 = {safe_divide(10, 2)}")  # 5.0
      print(f"10 / 0 = {safe_divide(10, 0)}")  # 0.0（捕获异常）

      # ========== 2. 异常处理 + 资源清理 ==========
      print("\n=== 异常处理 + 资源清理 ===")

      def read_config_file(filename: str) -> dict:
          """
          读取配置文件，确保文件被正确关闭
          """
          config = {}
          file = None  # 初始化变量，确保 finally 中可以访问
          
          try:
              # 打开文件
              file = open(filename, 'r')
              # 模拟读取配置（这里简化处理）
              content = file.read()
              print(f"✅ 成功读取文件: {filename}")
              config = {"content": content}
          except FileNotFoundError:
              # 文件不存在
              print(f"❌ 错误：文件 {filename} 不存在")
          except PermissionError:
              # 权限不足
              print(f"❌ 错误：没有权限读取文件 {filename}")
          except Exception as e:
              # 捕获其他所有异常
              print(f"❌ 未知错误: {e}")
          finally:
              # 无论是否发生异常，都会执行
              if file:
                  file.close()
                  print(f"✅ 文件 {filename} 已关闭")
          
          return config

      # 测试文件读取
      result = read_config_file("config.txt")  # 文件可能不存在

      # ========== 3. 异常链和重新抛出 ==========
      print("\n=== 异常链和重新抛出 ===")

      def process_data(data: dict) -> str:
          """
          处理数据，可能抛出多种异常
          """
          try:
              # 访问嵌套字典
              value = data['config']['database']['port']
              return f"端口: {value}"
          except KeyError as e:
              # 捕获 KeyError，添加更多上下文信息
              raise ValueError(f"配置缺失: {e}") from e
          except TypeError as e:
              # 捕获 TypeError（如 data 不是字典）
              raise ValueError(f"数据类型错误: {e}") from e

      # 测试异常处理
      try:
          result = process_data({"config": {"database": {"port": 3306}}})
          print(f"✅ {result}")
      except ValueError as e:
          print(f"❌ 处理失败: {e}")

      # ========== 4. 自定义异常 ==========
      print("\n=== 自定义异常 ===")

      class ServerError(Exception):
          """服务器相关异常"""
          pass

      class ConnectionTimeoutError(ServerError):
          """连接超时异常"""
          def __init__(self, host: str, port: int):
              self.host = host
              self.port = port
              super().__init__(f"连接 {host}:{port} 超时")

      def connect_server(host: str, port: int):
          """
          连接服务器，可能抛出自定义异常
          """
          # 模拟连接失败
          raise ConnectionTimeoutError(host, port)

      # 测试自定义异常
      try:
          connect_server("web-01", 8080)
      except ConnectionTimeoutError as e:
          print(f"❌ {e}")
          print(f"   主机: {e.host}, 端口: {e.port}")
      except ServerError as e:
          # 可以捕获父类异常
          print(f"❌ 服务器错误: {e}")
      ```
    </CodeGroup>

    **异常处理最佳实践**：

    * **具体异常优于通用异常**：`except KeyError:` 比 `except Exception:` 更精确
    * **使用 `finally` 清理资源**：确保文件、数据库连接等被正确关闭
    * **异常链**：使用 `raise ... from e` 保留原始异常信息
    * **自定义异常**：为特定业务场景定义专门的异常类

    **运行脚本**：

    ```bash theme={null}
    python 09_exception.py
    ```

    **验证步骤**：

    1. 脚本能正常运行，无语法错误
    2. 观察不同异常的处理方式
    3. 确认 `finally` 块总是执行
    4. 确认自定义异常能正确抛出和捕获
  </Step>
</Steps>

***

## <span style={{ display: 'inline-flex', alignItems: 'center', gap: '8px' }}><Icon icon="lightbulb" size={26} color="#059669" /> 拓展任务 (30 mins)</span>

<CardGroup cols={2}>
  <Card title="挑战 1：字典推导式进阶" icon="code">
    **任务**：从以下数据中，使用字典推导式提取所有 CPU 使用率超过 50% 的服务器，并计算平均 CPU 使用率。

    ```python theme={null}
    metrics = [
        {"host": "web-01", "cpu": 45, "mem": 60},
        {"host": "db-01",  "cpu": 85, "mem": 90},
        {"host": "web-02", "cpu": 10, "mem": 20},
        {"host": "cache-01", "cpu": 70, "mem": 40},
    ]
    ```

    **提示**：

    * 使用字典推导式筛选：`{k: v for item in list if condition}`
    * 计算平均值：`sum(values) / len(values)`
  </Card>

  <Card title="挑战 2：装饰器组合" icon="magic">
    **任务**：创建一个 `@timing` 装饰器，统计函数执行时间，并和 `@retry` 装饰器组合使用。

    **提示**：

    * 使用 `time.time()` 记录开始和结束时间
    * 装饰器可以叠加：`@timing` 和 `@retry` 可以同时使用
    * 思考：装饰器的执行顺序是什么？
  </Card>
</CardGroup>

***

## <span style={{ display: 'inline-flex', alignItems: 'center', gap: '8px' }}><Icon icon="file" size={26} color="#059669" /> 今日产出物</span>

* `09_dict_ops.py` - 字典操作示例
* `09_decorator.py` - 装饰器实现
* `09_exception.py` - 异常处理示例

***

## <span style={{ display: 'inline-flex', alignItems: 'center', gap: '8px' }}><Icon icon="github" size={26} color="#059669" /> 参考代码</span>

<CardGroup cols={2}>
  <Card title="查看参考代码" icon="code" href="https://github.com/akriamail/insightful-ops/tree/main/docs/mintlify/scripts/training/30daystudy/week-2/day-09">
    在 GitHub 查看完整的示例代码

    <br />

    <small>包含字典、装饰器和异常处理示例</small>
  </Card>

  <Card title="在线运行" icon="play" href="https://replit.com">
    使用在线编辑器测试代码

    <br />

    <small>无需本地环境配置</small>
  </Card>
</CardGroup>

***

## 实际应用场景

<CardGroup cols={2}>
  <Card title="字典在运维中的应用" icon="server">
    * **配置管理**：使用字典存储服务器配置、环境变量
    * **API 响应处理**：解析 JSON 格式的 API 响应（本质是字典）
    * **监控数据**：存储和查询服务器监控指标（CPU、内存、磁盘等）
    * **数据转换**：使用字典推导式快速筛选和转换数据
  </Card>

  <Card title="装饰器在运维中的应用" icon="code">
    * **重试机制**：网络请求、数据库连接失败自动重试
    * **性能监控**：统计脚本执行时间，定位性能瓶颈
    * **日志记录**：自动记录函数调用、参数和返回值
    * **权限校验**：检查用户是否有权限执行特定操作
    * **缓存机制**：缓存函数结果，避免重复计算
  </Card>
</CardGroup>

<Tip>
  **与 Day 10 的关联**：今天学习的字典操作和异常处理，明天会学习如何用 Python 处理文件（JSON、YAML、CSV）和正则表达式。字典是处理这些数据格式的基础。
</Tip>

***

<CardGroup cols={2}>
  <Card title="上一天: 基础补漏" href="/training/30daystudy/day-08">
    Day 08 | Python 环境与高频语法
  </Card>

  <Card title="下一天: 文件与正则" href="/training/30daystudy/day-10">
    Day 10 | 文件处理与正则实战
  </Card>
</CardGroup>