2 · 数据流转:变量、类型与内存初探
🔗 知识图谱导航:阅读本文前,建议先掌握/回顾 《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 像一张下拉菜单,告诉编辑器和读者:任务状态只应该从四个词里选。新手写项目时,先把边界写清楚,后面就不容易把 finished、complete 这类乱七八糟的状态混进来。
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.3 → False |
abs(a - b) < 1e-9 或 math.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.