Day 09 | 数据结构与装饰器

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

学习内容 (30 mins)

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

字典进阶 (15 mins)

什么是字典？字典（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}

装饰器 (10 mins)

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

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

装饰器原理：

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

装饰器应用场景：

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

异常处理 (5 mins)

什么是异常处理？异常处理是捕获和处理程序运行时的错误，防止程序崩溃。为什么必须正确处理异常？

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

异常处理规范：

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

代码任务 (90 mins)

环境准备

确保虚拟环境已激活，并安装必要的包：

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

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

任务 A：字典处理实战

编写 09_dict_ops.py，处理模拟的服务器监控数据。任务分解：

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

#!/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}

代码解释：

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

运行脚本：

python 09_dict_ops.py

验证步骤：

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

常见错误：

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

任务 B：编写重试装饰器

在运维脚本中，网络波动很常见。编写一个 @retry 装饰器，实现”报错自动重试 3 次”。任务分解：

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

#!/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}")

装饰器原理解析：

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

运行脚本：

python 09_decorator.py

验证步骤：

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

常见错误：

❌ 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__' - 可能是装饰器结构错误

任务 C：异常处理实战

编写 09_exception.py，演示正确的异常处理方式。

#!/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}")

异常处理最佳实践：

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

运行脚本：

python 09_exception.py

验证步骤：

脚本能正常运行，无语法错误
观察不同异常的处理方式
确认 finally 块总是执行
确认自定义异常能正确抛出和捕获

拓展任务 (30 mins)

挑战 1：字典推导式进阶

任务：从以下数据中，使用字典推导式提取所有 CPU 使用率超过 50% 的服务器，并计算平均 CPU 使用率。

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)

挑战 2：装饰器组合

任务：创建一个 @timing 装饰器，统计函数执行时间，并和 @retry 装饰器组合使用。提示：

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

今日产出物

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

参考代码

查看参考代码

在 GitHub 查看完整的示例代码

在线运行

使用在线编辑器测试代码

实际应用场景

字典在运维中的应用

配置管理：使用字典存储服务器配置、环境变量
API 响应处理：解析 JSON 格式的 API 响应（本质是字典）
监控数据：存储和查询服务器监控指标（CPU、内存、磁盘等）
数据转换：使用字典推导式快速筛选和转换数据

装饰器在运维中的应用

重试机制：网络请求、数据库连接失败自动重试
性能监控：统计脚本执行时间，定位性能瓶颈
日志记录：自动记录函数调用、参数和返回值
权限校验：检查用户是否有权限执行特定操作
缓存机制：缓存函数结果，避免重复计算

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

上一天: 基础补漏

Day 08 | Python 环境与高频语法

下一天: 文件与正则

Day 10 | 文件处理与正则实战

网站概览 | Overview

项目愿景 | Project Vision

内部培训 | Internal Training

技术规范 | Protocols

基础设施 | Infrastructure

工作安排 | Work Plan

团队协作 | Team Collab

Day 09 | 数据结构与装饰器

学习内容 (30 mins)

代码任务 (90 mins)

拓展任务 (30 mins)

挑战 1：字典推导式进阶

挑战 2：装饰器组合

今日产出物

参考代码

查看参考代码

在线运行

实际应用场景

字典在运维中的应用

装饰器在运维中的应用

上一天: 基础补漏

下一天: 文件与正则

网站概览 | Overview

项目愿景 | Project Vision

内部培训 | Internal Training

技术规范 | Protocols

基础设施 | Infrastructure

工作安排 | Work Plan

团队协作 | Team Collab

​ 学习内容 (30 mins)

​ 代码任务 (90 mins)

​ 拓展任务 (30 mins)

挑战 1：字典推导式进阶

挑战 2：装饰器组合

​ 今日产出物

​ 参考代码

查看参考代码

在线运行

​实际应用场景

字典在运维中的应用

装饰器在运维中的应用

上一天: 基础补漏

下一天: 文件与正则

学习内容 (30 mins)

代码任务 (90 mins)

拓展任务 (30 mins)

今日产出物

参考代码

实际应用场景