文章

2 · 数据流转:变量、类型与内存初探

#002 · 2026-04-16 · Python

🔗 知识图谱导航:阅读本文前,建议先掌握/回顾 《1 · 环境、终端运行与第一个脚本》 中的核心概念;本文会在这个基础上继续推进。 承上启下:上一篇我们跑通了第一个 CLI 工具。现在深入一层——Python 里的变量到底是什么?为什么 a = 1; b = a; a = 2 之后 b 还是 1

极客解析:先把数据流、控制流和模块边界跑通,再谈抽象;每段代码都围绕一个可执行 CLI 闭环展开。


痛点与架构

Python 变量不是"盒子",而是标签(引用)


int a = 1;   ← a 是一个装着 1 的盒子

# 正确心智模型(Python)
a = 1        ← a 是贴在对象 1 上的标签
b = a        ← b 也贴在同一个对象 1 上
a = 2        ← a 标签撕下来贴到对象 2 上,b 不受影响
类型 可变性 示例 内存行为
int / float / str / bool 不可变 x = 42 修改 = 创建新对象
list / dict / set 可变 lst = [1,2] 修改 = 原地改变
tuple / frozenset 不可变 t = (1,2) 修改 = 报错

实战演练场

任务状态追踪器演示所有类型特性,代码完全自包含,无需任何外部文件:

步步为营:核心逻辑自适应拆解

导师提示:这一篇的目标是让小白真正看懂“变量、类型、列表、字典、集合”怎么在一个脚本里配合工作。不要先背概念,先运行每个小片段,看数据怎么从字符串、字典、列表一路变成终端看板。

Step 1:用类型别名给任务状态划边界

核心源码(逐字来自文末完整源码)

TaskStatus = Literal["pending", "running", "done", "failed"]
TaskID = int

可运行演示(补齐 Mock 数据与 print 反馈)

from typing import Literal

# Literal 像给变量划定“只能选这几个按钮”。
TaskStatus = Literal["pending", "running", "done", "failed"]
TaskID = int

status: TaskStatus = "running"
task_id: TaskID = 2
print(f"任务编号:{task_id}")
print(f"当前状态:{status}")
print("允许的状态只有:pending / running / done / failed")

大白话解析:这一步先建立“类型提示只是提示,不是枷锁”的心智。TaskStatus 像一张下拉菜单,告诉编辑器和读者:任务状态只应该从四个词里选。新手写项目时,先把边界写清楚,后面就不容易把 finishedcomplete 这类乱七八糟的状态混进来。

Step 2:看懂变量是标签,不是盒子

核心源码(逐字来自文末完整源码)

def demo_immutable_vs_mutable() -> None:
    """演示可变与不可变对象的内存行为差异。"""
    print("\n  ── 内存模型演示 ──────────────────────────")

    # 不可变:int
    a: int = 100
    b: int = a
    a = 200
    print(f"  int:   a={a}, b={b}  ← b 不受影响(不可变)")

    # 可变:list
    lst_a: list[int] = [1, 2, 3]
    lst_b: list[int] = lst_a        # 同一对象!
    lst_a.append(4)
    print(f"  list:  lst_a={lst_a}, lst_b={lst_b}  ← 共享引用(可变)")

    # 正确复制方式
    lst_c: list[int] = lst_a.copy()
    lst_a.append(5)
    print(f"  copy:  lst_a={lst_a}, lst_c={lst_c}  ← 独立副本")

    # id() 查看内存地址
    x: str = "geek"
    y: str = "geek"
    print(f"  str interning: id(x)==id(y) → {id(x) == id(y)}  ← 小字符串复用")

可运行演示(补齐 Mock 数据与 print 反馈)

from datetime import datetime, timedelta
from typing import Literal

# ── 类型别名(Python 3.12 推荐写法)──────────────────────────

def demo_immutable_vs_mutable() -> None:
    """演示可变与不可变对象的内存行为差异。"""
    print("\n  ── 内存模型演示 ──────────────────────────")

    # 不可变:int
    a: int = 100
    b: int = a
    a = 200
    print(f"  int:   a={a}, b={b}  ← b 不受影响(不可变)")

    # 可变:list
    lst_a: list[int] = [1, 2, 3]
    lst_b: list[int] = lst_a        # 同一对象!
    lst_a.append(4)
    print(f"  list:  lst_a={lst_a}, lst_b={lst_b}  ← 共享引用(可变)")

    # 正确复制方式
    lst_c: list[int] = lst_a.copy()
    lst_a.append(5)
    print(f"  copy:  lst_a={lst_a}, lst_c={lst_c}  ← 独立副本")

    # id() 查看内存地址
    x: str = "geek"
    y: str = "geek"
    print(f"  str interning: id(x)==id(y) → {id(x) == id(y)}  ← 小字符串复用")

# 单独运行这个函数,直接观察 int、list、copy 的区别。
print("开始演示:变量其实是贴在对象上的标签")
demo_immutable_vs_mutable()

大白话解析:Python 变量更像便利贴,不像盒子。b = a 是把第二张便利贴也贴到同一个对象上;列表是可变对象,所以 lst_a.append(4) 会让 lst_b 也看到变化。copy() 就像复印一份清单,后续再改原件,复印件不会跟着动。

Step 3:用列表套字典描述真实任务

核心源码(逐字来自文末完整源码)

def build_task_graph() -> list[dict]:
    """构建一个模拟任务图谱,演示字典与列表的组合使用。"""
    now = datetime.now()
    tasks: list[dict] = [
        {
            "id":       1,
            "name":     "数据采集",
            "status":   "done",
            "priority": 1,
            "eta":      now - timedelta(hours=2),
            "tags":     {"crawler", "io"},
        },
        {
            "id":       2,
            "name":     "文本清洗",
            "status":   "running",
            "priority": 2,
            "eta":      now + timedelta(minutes=30),
            "tags":     {"nlp", "transform"},
        },
        {
            "id":       3,
            "name":     "向量入库",
            "status":   "pending",
            "priority": 3,
            "eta":      now + timedelta(hours=1),
            "tags":     {"vector", "db"},
        },
        {
            "id":       4,
            "name":     "模型推理",
            "status":   "failed",
            "priority": 2,
            "eta":      now - timedelta(minutes=10),
            "tags":     {"ai", "inference"},
        },
    ]
    return tasks

可运行演示(补齐 Mock 数据与 print 反馈)

from datetime import datetime, timedelta

def build_task_graph() -> list[dict]:
    """构建一个模拟任务图谱,演示字典与列表的组合使用。"""
    now = datetime.now()
    tasks: list[dict] = [
        {
            "id":       1,
            "name":     "数据采集",
            "status":   "done",
            "priority": 1,
            "eta":      now - timedelta(hours=2),
            "tags":     {"crawler", "io"},
        },
        {
            "id":       2,
            "name":     "文本清洗",
            "status":   "running",
            "priority": 2,
            "eta":      now + timedelta(minutes=30),
            "tags":     {"nlp", "transform"},
        },
        {
            "id":       3,
            "name":     "向量入库",
            "status":   "pending",
            "priority": 3,
            "eta":      now + timedelta(hours=1),
            "tags":     {"vector", "db"},
        },
        {
            "id":       4,
            "name":     "模型推理",
            "status":   "failed",
            "priority": 2,
            "eta":      now - timedelta(minutes=10),
            "tags":     {"ai", "inference"},
        },
    ]
    return tasks

tasks = build_task_graph()
print(f"一共生成 {len(tasks)} 个任务")
first = tasks[0]
print(f"第一个任务:{first['name']} / 状态={first['status']} / 标签={sorted(first['tags'])}")

大白话解析:任务图谱的本质很朴素:列表负责“有多条任务”,字典负责“每条任务有哪些字段”。你可以把它想成 Excel:每一行是一个任务,每一列是 id、name、status、priority、tags。

Step 4:用状态图标把数据翻译成人话

核心源码(逐字来自文末完整源码)

STATUS_ICON: dict[TaskStatus, str] = {
    "pending": "⏳",
    "running": "🔄",
    "done":    "✅",
    "failed":  "❌",
}

可运行演示(补齐 Mock 数据与 print 反馈)

from typing import Literal

TaskStatus = Literal["pending", "running", "done", "failed"]
STATUS_ICON: dict[TaskStatus, str] = {
    "pending": "⏳",
    "running": "🔄",
    "done":    "✅",
    "failed":  "❌",
}

for status, icon in STATUS_ICON.items():
    print(f"{status:<8} => {icon}")

大白话解析STATUS_ICON 像翻译词典:机器读的是 running,人更容易一眼看懂 🔄。做命令行工具时,适量图标能帮新手快速定位状态,但真实业务逻辑仍然要保存稳定的英文状态值。

Step 5:把任务列表打印成状态看板

核心源码(逐字来自文末完整源码)

def print_task_board(tasks: list[dict]) -> None:
    """以看板形式打印任务状态。"""
    width = 56
    print(f"\n  ┌{'─' * width}┐")
    print(f"  │{'  📋 任务图谱状态看板':^{width}}│")
    print(f"  ├{'─' * 4}┬{'─' * 16}┬{'─' * 10}┬{'─' * 6}┬{'─' * 18}┤")
    print(f"  │ ID │ {'任务名':<14} │ {'状态':<8} │ P{'':4} │ {'预计时间':<16} │")
    print(f"  ├{'─' * 4}┼{'─' * 16}┼{'─' * 10}┼{'─' * 6}┼{'─' * 18}┤")
    for t in tasks:
        icon = STATUS_ICON.get(t["status"], "?")
        eta_str = t["eta"].strftime("%H:%M")
        print(
            f"  │ {t['id']:>2} │ {t['name']:<14} │ "
            f"{icon} {t['status']:<6} │ P{t['priority']}{'':4} │ "
            f"{eta_str:<16} │"
        )
    print(f"  └{'─' * 4}┴{'─' * 16}┴{'─' * 10}┴{'─' * 6}┴{'─' * 18}┘")

    # 类型操作演示:集合运算
    all_tags: set[str] = set()
    for t in tasks:
        all_tags |= t["tags"]          # 集合合并(|= 运算符)
    running_tags: set[str] = {
        tag for t in tasks if t["status"] == "running"
        for tag in t["tags"]
    }
    print(f"\n  全部标签: {sorted(all_tags)}")
    print(f"  运行中标签: {sorted(running_tags)}")

    # 统计各状态数量(字典推导式)
    status_count: dict[str, int] = {}
    for t in tasks:
        status_count[t["status"]] = status_count.get(t["status"], 0) + 1
    print(f"\n  状态统计: {status_count}")

可运行演示(补齐 Mock 数据与 print 反馈)

from datetime import datetime, timedelta
from typing import Literal

TaskStatus = Literal["pending", "running", "done", "failed"]
STATUS_ICON: dict[TaskStatus, str] = {
    "pending": "⏳",
    "running": "🔄",
    "done":    "✅",
    "failed":  "❌",
}

def print_task_board(tasks: list[dict]) -> None:
    """以看板形式打印任务状态。"""
    width = 56
    print(f"\n{'─' * width}┐")
    print(f"  │{'  📋 任务图谱状态看板':^{width}}│")
    print(f"  ├{'─' * 4}{'─' * 16}{'─' * 10}{'─' * 6}{'─' * 18}┤")
    print(f"  │ ID │ {'任务名':<14}{'状态':<8} │ P{'':4}{'预计时间':<16} │")
    print(f"  ├{'─' * 4}{'─' * 16}{'─' * 10}{'─' * 6}{'─' * 18}┤")
    for t in tasks:
        icon = STATUS_ICON.get(t["status"], "?")
        eta_str = t["eta"].strftime("%H:%M")
        print(
            f"  │ {t['id']:>2}{t['name']:<14} │ "
            f"{icon} {t['status']:<6} │ P{t['priority']}{'':4} │ "
            f"{eta_str:<16} │"
        )
    print(f"  └{'─' * 4}{'─' * 16}{'─' * 10}{'─' * 6}{'─' * 18}┘")

    # 类型操作演示:集合运算
    all_tags: set[str] = set()
    for t in tasks:
        all_tags |= t["tags"]          # 集合合并(|= 运算符)
    running_tags: set[str] = {
        tag for t in tasks if t["status"] == "running"
        for tag in t["tags"]
    }
    print(f"\n  全部标签: {sorted(all_tags)}")
    print(f"  运行中标签: {sorted(running_tags)}")

    # 统计各状态数量(字典推导式)
    status_count: dict[str, int] = {}
    for t in tasks:
        status_count[t["status"]] = status_count.get(t["status"], 0) + 1
    print(f"\n  状态统计: {status_count}")

now = datetime.now()
tasks = [
    {"id": 1, "name": "数据采集", "status": "done", "priority": 1, "eta": now, "tags": {"crawler", "io"}},
    {"id": 2, "name": "文本清洗", "status": "running", "priority": 2, "eta": now, "tags": {"nlp", "transform"}},
]
print_task_board(tasks)

大白话解析:看板输出的目标不是炫酷,而是把一堆字典变成能读的表格。for t in tasks 像逐行扫描 Excel,集合运算 |= 像把所有标签倒进同一个篮子里去重,最后统计每种状态有几条。

Step 6:把字符串安全转换成具体类型

核心源码(逐字来自文末完整源码)

def demo_type_conversion() -> None:
    """演示类型转换与常见陷阱。"""
    print("\n  ── 类型转换演示 ──────────────────────────")

    # 安全转换
    raw_inputs: list[str] = ["42", "3.14", "true", "100"]
    for raw in raw_inputs:
        try:
            if "." in raw:
                val: float = float(raw)
                print(f"  '{raw}' → float: {val}")
            elif raw.lower() in ("true", "false"):
                val_b: bool = raw.lower() == "true"
                print(f"  '{raw}' → bool:  {val_b}")
            else:
                val_i: int = int(raw)
                print(f"  '{raw}' → int:   {val_i}")
        except ValueError as e:
            print(f"  '{raw}' → 转换失败: {e}")

    # 浮点精度陷阱
    result: float = 0.1 + 0.2
    print(f"\n  0.1 + 0.2 = {result}  ← 浮点精度问题")
    print(f"  round(0.1+0.2, 1) = {round(result, 1)}  ← 用 round() 修正")

可运行演示(补齐 Mock 数据与 print 反馈)

def demo_type_conversion() -> None:
    """演示类型转换与常见陷阱。"""
    print("\n  ── 类型转换演示 ──────────────────────────")

    # 安全转换
    raw_inputs: list[str] = ["42", "3.14", "true", "100"]
    for raw in raw_inputs:
        try:
            if "." in raw:
                val: float = float(raw)
                print(f"  '{raw}' → float: {val}")
            elif raw.lower() in ("true", "false"):
                val_b: bool = raw.lower() == "true"
                print(f"  '{raw}' → bool:  {val_b}")
            else:
                val_i: int = int(raw)
                print(f"  '{raw}' → int:   {val_i}")
        except ValueError as e:
            print(f"  '{raw}' → 转换失败: {e}")

    # 浮点精度陷阱
    result: float = 0.1 + 0.2
    print(f"\n  0.1 + 0.2 = {result}  ← 浮点精度问题")
    print(f"  round(0.1+0.2, 1) = {round(result, 1)}  ← 用 round() 修正")

# 单独运行类型转换演示,看看字符串如何变成 int / float / bool。
demo_type_conversion()

大白话解析:终端参数、表单输入、文件内容刚读进来时大多都是字符串。类型转换就像把快递包裹拆开重新归类:"42" 变整数,"3.14" 变小数,"true" 变布尔值。try/except 是安全网,防止坏输入把程序砸停。

极客实战:完整源码与运行

现在,把上面的积木拼起来,将以下完整代码放进你的编辑器,运行它。先看整体闭环,再回头逐段改参数,你会更容易建立工程直觉。

# task_tracker.py
"""
任务状态追踪器 —— 演示 Python 类型系统与内存模型。
用法:python3 task_tracker.py
"""

import argparse
import sys
from datetime import datetime, timedelta
from typing import Literal

# ── 类型别名(Python 3.12 推荐写法)──────────────────────────
TaskStatus = Literal["pending", "running", "done", "failed"]
TaskID = int


def demo_immutable_vs_mutable() -> None:
    """演示可变与不可变对象的内存行为差异。"""
    print("\n  ── 内存模型演示 ──────────────────────────")

    # 不可变:int
    a: int = 100
    b: int = a
    a = 200
    print(f"  int:   a={a}, b={b}  ← b 不受影响(不可变)")

    # 可变:list
    lst_a: list[int] = [1, 2, 3]
    lst_b: list[int] = lst_a        # 同一对象!
    lst_a.append(4)
    print(f"  list:  lst_a={lst_a}, lst_b={lst_b}  ← 共享引用(可变)")

    # 正确复制方式
    lst_c: list[int] = lst_a.copy()
    lst_a.append(5)
    print(f"  copy:  lst_a={lst_a}, lst_c={lst_c}  ← 独立副本")

    # id() 查看内存地址
    x: str = "geek"
    y: str = "geek"
    print(f"  str interning: id(x)==id(y) → {id(x) == id(y)}  ← 小字符串复用")


def build_task_graph() -> list[dict]:
    """构建一个模拟任务图谱,演示字典与列表的组合使用。"""
    now = datetime.now()
    tasks: list[dict] = [
        {
            "id":       1,
            "name":     "数据采集",
            "status":   "done",
            "priority": 1,
            "eta":      now - timedelta(hours=2),
            "tags":     {"crawler", "io"},
        },
        {
            "id":       2,
            "name":     "文本清洗",
            "status":   "running",
            "priority": 2,
            "eta":      now + timedelta(minutes=30),
            "tags":     {"nlp", "transform"},
        },
        {
            "id":       3,
            "name":     "向量入库",
            "status":   "pending",
            "priority": 3,
            "eta":      now + timedelta(hours=1),
            "tags":     {"vector", "db"},
        },
        {
            "id":       4,
            "name":     "模型推理",
            "status":   "failed",
            "priority": 2,
            "eta":      now - timedelta(minutes=10),
            "tags":     {"ai", "inference"},
        },
    ]
    return tasks


STATUS_ICON: dict[TaskStatus, str] = {
    "pending": "⏳",
    "running": "🔄",
    "done":    "✅",
    "failed":  "❌",
}


def print_task_board(tasks: list[dict]) -> None:
    """以看板形式打印任务状态。"""
    width = 56
    print(f"\n{'─' * width}┐")
    print(f"  │{'  📋 任务图谱状态看板':^{width}}│")
    print(f"  ├{'─' * 4}{'─' * 16}{'─' * 10}{'─' * 6}{'─' * 18}┤")
    print(f"  │ ID │ {'任务名':<14}{'状态':<8} │ P{'':4}{'预计时间':<16} │")
    print(f"  ├{'─' * 4}{'─' * 16}{'─' * 10}{'─' * 6}{'─' * 18}┤")
    for t in tasks:
        icon = STATUS_ICON.get(t["status"], "?")
        eta_str = t["eta"].strftime("%H:%M")
        print(
            f"  │ {t['id']:>2}{t['name']:<14} │ "
            f"{icon} {t['status']:<6} │ P{t['priority']}{'':4} │ "
            f"{eta_str:<16} │"
        )
    print(f"  └{'─' * 4}{'─' * 16}{'─' * 10}{'─' * 6}{'─' * 18}┘")

    # 类型操作演示:集合运算
    all_tags: set[str] = set()
    for t in tasks:
        all_tags |= t["tags"]          # 集合合并(|= 运算符)
    running_tags: set[str] = {
        tag for t in tasks if t["status"] == "running"
        for tag in t["tags"]
    }
    print(f"\n  全部标签: {sorted(all_tags)}")
    print(f"  运行中标签: {sorted(running_tags)}")

    # 统计各状态数量(字典推导式)
    status_count: dict[str, int] = {}
    for t in tasks:
        status_count[t["status"]] = status_count.get(t["status"], 0) + 1
    print(f"\n  状态统计: {status_count}")


def demo_type_conversion() -> None:
    """演示类型转换与常见陷阱。"""
    print("\n  ── 类型转换演示 ──────────────────────────")

    # 安全转换
    raw_inputs: list[str] = ["42", "3.14", "true", "100"]
    for raw in raw_inputs:
        try:
            if "." in raw:
                val: float = float(raw)
                print(f"  '{raw}' → float: {val}")
            elif raw.lower() in ("true", "false"):
                val_b: bool = raw.lower() == "true"
                print(f"  '{raw}' → bool:  {val_b}")
            else:
                val_i: int = int(raw)
                print(f"  '{raw}' → int:   {val_i}")
        except ValueError as e:
            print(f"  '{raw}' → 转换失败: {e}")

    # 浮点精度陷阱
    result: float = 0.1 + 0.2
    print(f"\n  0.1 + 0.2 = {result}  ← 浮点精度问题")
    print(f"  round(0.1+0.2, 1) = {round(result, 1)}  ← 用 round() 修正")


def main() -> None:
    parser = argparse.ArgumentParser(description="任务状态追踪器 & 类型系统演示")
    parser.add_argument(
        "--mode",
        choices=["board", "memory", "types", "all"],
        default="all",
        help="演示模式(默认: all)",
    )
    args = parser.parse_args()

    tasks = build_task_graph()

    if args.mode in ("board", "all"):
        print_task_board(tasks)
    if args.mode in ("memory", "all"):
        demo_immutable_vs_mutable()
    if args.mode in ("types", "all"):
        demo_type_conversion()


if __name__ == "__main__":
    main()

终端预期输出:

$ python3 task_tracker.py --mode board

  ┌────────────────────────────────────────────────────────┐
  │                   📋 任务图谱状态看板                    │
  ├────┬────────────────┬──────────┬──────┬──────────────────┤
  │ ID │ 任务名          │ 状态      │ P    │ 预计时间          │
  ├────┼────────────────┼──────────┼──────┼──────────────────┤
1 │ 数据采集        │ ✅ done   │ P1   │ 11:00            │
2 │ 文本清洗        │ 🔄 running│ P2   │ 13:30            │
3 │ 向量入库        │ ⏳ pending│ P3   │ 14:00            │
4 │ 模型推理        │ ❌ failed │ P2   │ 12:50            │
  └────┴────────────────┴──────────┴──────┴──────────────────┘

  全部标签: ['ai', 'crawler', 'db', 'inference', 'io', 'nlp', 'transform', 'vector']
  运行中标签: ['nlp', 'transform']
  状态统计: {'done': 1, 'running': 1, 'pending': 1, 'failed': 1}

避坑指南

示例 正确做法
可变默认参数 def f(lst=[]) 多次调用共享同一列表 def f(lst=None): lst = lst or []
is vs == a is b 比较对象身份,不是值 比较值用 ==,判断 None 用 is None
整除结果 7 / 2 = 3.5(不是 3) 需要整数用 7 // 2
浮点比较 0.1 + 0.2 == 0.3False abs(a - b) < 1e-9math.isclose

NexDo Time ⚡

5 分钟极客微操:给 task_tracker.py 增加 --filter STATUS 参数,只显示指定状态的任务(如 --filter running)。提示:用列表推导式 [t for t in tasks if t["status"] == args.filter] 过滤。

Don’t wait for next time, do it in the next moment.