跳转至

第 1 周(9/8 二 – 9/13 日):语言进阶——把 Python 的"机制"学透

本周目标:迭代器/生成器、闭包/装饰器、上下文管理器、数据模型(魔术方法)、dataclass/Enum、OOP 进阶(组合、ABC、Protocol)、异常进阶与 3.12–3.14 新语法。周日晚交付小项目 文本处理管线

时长:周二–周五 4h × 4 + 周六日 10h × 2 = 36h。工作日只做"1 复习 + 1 新块 + 1 练习";周末开大块。

本周练习exercises/week1/w1_0N_*.py(骨架)+ 同目录 test_w1_0N_*.py(已给测试)。运行:uv run pytest exercises/week1/test_w1_01_generators.py -q。答案在 exercises/solutions/week1/(做完再看)。

对应 learn-agent:M1 → 流式输出;M2 → Day 3 的 @tool 装饰器与 inspect;M4 → 工具/消息/记忆的领域模型与 Protocol 接口;M5 → 工具错误如何变成可读文本。


周二 9/8(4h)|M1a 迭代协议与生成器

块 1(复习):Day 1–3 回忆测验 15 分钟(checklist.md A 档表随机抽 10 条默写);收尾周一项目遗留。

块 2(新知识): | 知识点 | 档 | | --- | --- | | 可迭代对象 vs 迭代器:iter()/next()StopIterationfor 的真实展开 | A | | 生成器函数 yield:惰性、状态保存、只能遍历一次;生成器表达式 (x for x in ...) | A | | 生成器管线:lines → parse → filter → aggregate,每级只处理一条 | A | | 用生成器读大文件(逐行/分块)而不 read() 全部 | A | | return 在生成器里的含义(变成 StopIteration.value)、yield from 委托 | B | | send()/throw()/close()、经典协程 | C |

预测(写在纸上,再跑):

def gen():
    print("start"); yield 1; print("middle"); yield 2; print("end")
g = gen()
print(type(g)); print(next(g)); print(next(g))
print(list(g), list(g))          # 第二个 list 是什么?
squares = (x * x for x in range(3))
print(sum(squares), sum(squares))  # 第二个 sum 是什么?

块 3(练习)w1_01_generators.py

函数 行为 测试要点
countdown(n: int) -> Iterator[int] 生成器,n…0 list(countdown(3)) == [3,2,1,0];返回值是生成器(inspect.isgenerator
read_chunks(path: Path, size: int) -> Iterator[str] 逐块读取文本(UTF-8) 临时文件;块拼起来等于原文
parse_records(lines: Iterable[str]) -> Iterator[dict] 每行 name,score,跳过空行与注释行 #score 转 int,非法行 raise ValueError(带行号) 惰性:传入无限生成器也能取前 2 条
take(n: int, it: Iterable[T]) -> list[T] 取前 n 个(用 itertools.islice 对无限生成器有效
running_mean(nums: Iterable[float]) -> Iterator[float] 逐个产出到目前为止的平均值 [2, 4, 6] → [2.0, 3.0, 4.0]
flatten(nested: Iterable) -> Iterator 递归展平任意嵌套列表/元组(字符串不拆),用 yield from [1,[2,[3,"ab"]]] → [1,2,3,"ab"]
class Countdown 手写迭代器类:__iter__ 返回 self__next__StopIteration 可被 for 遍历;遍历一次后耗尽

收尾:日志;明日预览 itertools 文档首页 5 分钟。


周三 9/9(4h)|M1b itertools 与惰性思维

块 1(复习):默写生成器管线三段代码;重做昨天最慢的一题。

块 2(新知识): | 知识点 | 档 | | --- | --- | | itertoolschain/islice/count/cycle/repeat/accumulate/batched(3.12)/pairwise(3.10)/groupby/takewhile/dropwhile/zip_longest/product/permutations/combinations/starmap | A(前 8 个)/ B(其余) | | groupby 必须先排序的坑;batched 替代手写分块 | A | | enumerate/zip/reversed/sorted/any/all/min/max/sum 与生成器搭配;map/filter 返回惰性迭代器 | A | | 内存对比实验:sys.getsizeof([x for x in range(10**6)]) vs 生成器;tracemalloc 看峰值 | B | | more-itertools 第三方库(知道存在) | C |

练习w1_02_itertools.py

函数 行为
chunked(it, size) -> Iterator[tuple] itertools.batched 实现;size<=0ValueError
window(it, n) -> Iterator[tuple] 滑动窗口(n=2 时等价 pairwise);用 deque(maxlen=n)
group_by_key(records, key) -> dict[str, list] sorted(key=)groupby,返回普通字典
interleave(*its) -> Iterator 交错合并,长度不同用 zip_longest 跳过填充值
first_true(it, pred, default=None) 第一个满足条件的元素
cumulative_max(nums) -> list accumulate(nums, max)
top_k_pairs(words, k) -> list[tuple[str,int]] 词频 Counter + most_common
memory_report() -> dict 返回 {"list_bytes": ..., "gen_bytes": ...},验证生成器对象远小于列表

小实验(块 3 末 15 分钟):写一个 300 万行的临时文件,对比 for line in ff.readlines() 的峰值内存(tracemalloc.get_traced_memory()),把数字写进日志。


周四 9/10(4h)|M2a 闭包与装饰器基础

块 2(新知识): | 知识点 | 档 | | --- | --- | | 函数是对象:赋值、作参数、作返回值、__name__/__doc__/__module__ | A | | 闭包与 nonlocal晚绑定陷阱(循环里创建 lambda) | A | | 装饰器 = 接收函数返回函数的语法糖;@decorator 等价于 f = decorator(f) | A | | functools.wraps 保留元数据(不写会丢 __name__/__doc__,影响文档与调试) | A | | 经典三件套:计时、日志、重试(带次数与延迟) | A | | 装饰器的执行时机(导入时执行外层、调用时执行内层) | A |

预测

def outer():
    x = 1
    def inner():
        nonlocal x; x += 1; return x
    return inner
f = outer(); print(f(), f()); g = outer(); print(g())
fs = [lambda: i for i in range(3)]; print([h() for h in fs])          # 晚绑定
fs = [lambda i=i: i for i in range(3)]; print([h() for h in fs])      # 修复
def deco(fn):
    print("decorating", fn.__name__)
    return fn
@deco
def hello(): pass
print("module loaded")     # 顺序?

练习w1_03_decorators.py

装饰器/函数 行为 测试要点
make_counter() -> Callable[[], int] 闭包计数器,每次调用 +1 两个计数器独立
@timer 打印/记录耗时,返回原结果;用 wraps 被装饰函数 __name__ 不变
@log_calls(logger_list: list) f(args, kwargs) -> result 追加到列表(便于测试) 记录格式 "add(2, 3) -> 5"
@retry(times=3, exceptions=(ValueError,), delay=0.0) 失败重试,最后一次仍失败则重新抛出 用一个"前两次失败第三次成功"的桩函数验证调用次数
@validate_positive 所有位置参数必须 > 0,否则 ValueError
@once 只执行一次,之后返回缓存结果
compose(*fns) compose(f, g)(x) == f(g(x))

周五 9/11(4h)|M2b 带参装饰器、functools、注册表与 inspect

块 2(新知识): | 知识点 | 档 | | --- | --- | | 带参数的装饰器(三层)与"可选参数装饰器"写法 | A | | 类装饰器与用类实现装饰器(__call__) | B | | functools.cache/lru_cache(maxsize)(斐波那契对比)、partialsingledispatchcached_property | A / B | | 注册表模式@register 把函数放进 dict[str, Callable],按名字调用——插件、命令、(之后的)工具都是它 | A | | inspect.signature(fn):参数名、默认值、kindinspect.getdoctyping.get_type_hints(fn) | A | | 3.14 注解延迟求值(PEP 649):fn.__annotations__ 何时求值;annotationlib.get_annotations(fn, format=...) | B | | 用签名 + 类型提示自动生成"参数说明表"(字典)——这是从函数生成 JSON Schema 的 Python 部分 | A |

练习w1_04_registry_inspect.py

成员 行为
class Registry register(name=None) 装饰器(可带/不带括号);get(name)names();重名 → ValueError
@repeat(n) 重复调用 n 次并返回结果列表
@cached_fib:用 functools.cache 实现 fib(n) fib(80) 瞬间完成;对比无缓存版本调用次数
describe(fn) -> dict inspect.signature + get_type_hints 返回 {"name", "doc", "params": [{"name","type","default","required"}], "returns"}
call_with_kwargs(fn, data: dict) 按签名从字典挑出参数调用;缺必填参数 → TypeError(含缺失名);多余键忽略
to_str(value)singledispatch int→"整数 3"list→"列表(2 项)"、默认 repr

周末预览:读 contextlib 文档 10 分钟。


周六 9/12(10h)|M3 上下文管理器 + 函数式定位 + M4a 数据模型

上午(3 块)— M3: | 知识点 | 档 | | --- | --- | | with 的展开:__enter__ 返回值绑定 as__exit__(exc_type, exc, tb) 返回 True 吞异常 | A | | @contextlib.contextmanageryield 前后即 enter/exit;try/finally 保证清理 | A | | ExitStack 管理数量不定的资源;suppressredirect_stdoutclosingnullcontext | B | | 多个上下文 with a, b:;3.10 括号写法 | A | | 异步上下文管理器 async with(下周三讲 asyncio 时再用) | C(预告) | | 函数式工具定位lambda 只用于一行;map/filter 能读、写时优先推导式;reduce 只在"没有内建聚合"时用;operator.itemgetter/attrgetter/methodcallersorted(key=)min/max(key=) | A(会读)/ B(会写) |

练习 w1_05_context_managers.pyTimer(类实现,记录 elapsed)、@contextmanager temp_env(**vars)(临时设置环境变量并恢复)、@contextmanager cd(path)(临时切换目录并恢复,即使出错)、open_all(paths) -> list[TextIO]ExitStack)、suppress_and_log(exc_type, log: list)(吞掉指定异常并记录)、atomic_write(path)(写临时文件成功后 replace,失败不留半成品)。 练习 w1_06_functional.py:用 sorted/key/itemgetter 多字段排序、reduce 实现 composemap/filter 与推导式互写各一例(测试只验证结果,注释里写"你更愿意读哪种")、partial 预填参数、methodcaller

下午(3 块)— M4a 数据模型与魔术方法: | 知识点 | 档 | | --- | --- | | __repr__(给开发者,能 eval 回来最好)vs __str__(给用户);容器里显示用 repr | A | | __eq__ 必配 __hash__(定义 __eq__ 不定义 __hash__ → 不可哈希);相等与身份 | A | | 排序:__lt__ + functools.total_ordering | A | | 序列协议:__len__/__getitem__/__contains__/__iter__/__reversed__;实现 __getitem__for/in 自动可用 | A | | __bool____call__、算术 __add__/__radd__/__iadd____format__ | B | | __getattr__(找不到才调用)vs __getattribute__(每次都调用,危险) | B | | 官方 Data Model 文档;描述符与元类 | C |

预测:

class P:
    def __init__(self, x): self.x = x
    def __eq__(self, o): return self.x == o.x
p = P(1); print(p == P(1), p is P(1))
# print({p})            → TypeError: unhashable type: 'P'  为什么?
class V:
    def __init__(self, *xs): self.xs = list(xs)
    def __len__(self): return len(self.xs)
    def __getitem__(self, i): return self.xs[i]
v = V(1, 2, 3); print(len(v), v[0], 2 in v, list(reversed(v)), bool(V()))

练习 w1_07_data_model.pyclass Money(amount: Decimal, currency: str)__repr__/__str__/__eq__/__hash__/__lt__(total_ordering)/__add__/__radd__/__bool__,不同货币相加 → ValueError,可作 dict 键、可 sum()(用 __radd__ 处理 0);class Playlist:序列协议 + __contains__ + __iter__ + 切片返回新 Playlistclass Config__getattr__ 把字典键当属性读、缺失 → AttributeErrorclass Celsius__format__ 支持 f"{t:.1f}"f"{t:F}"(转华氏)。

晚上(2 块)— M4b dataclassEnum 进阶: | 知识点 | 档 | | --- | --- | | @dataclass 选项:frozen(不可变 + 可哈希)、slots(省内存、防拼错属性)、kw_onlyordereq | A | | field(default_factory, repr=False, compare=False, kw_only=True)__post_init__ 校验、InitVar | A / B | | dataclasses.asdict/astuple/replace/fields、3.13 copy.replace() | B | | Enum/StrEnum(3.11)/IntEnum/Flag/auto、按值/名查找、Enum 做状态机与 match 搭配 | A | | typing.NamedTuple vs dataclass 选择 | B | | 什么时候用 dataclass、什么时候用下周的 Pydantic(边界:内部领域模型 vs 外部输入校验) | A |

练习 w1_08_dataclass_enum.pyPriority(IntEnum)Status(StrEnum)@dataclass(frozen=True, slots=True) Task(title, priority=Priority.NORMAL, tags=field(default_factory=list))__post_init__ 校验非空标题、with_status(new) -> Taskreplace)、sort_tasks(tasks)(按优先级降序、标题升序)、to_dict(task) -> dictasdict + 枚举转值)、from_dict(d) -> TaskWeekday(Flag) 组合与判断、describe(status)match 枚举成员)。


周日 9/13(10h)|M4c OOP 进阶 + M5 异常与新语法 + 小项目 + 周测

上午(3 块)— M4c OOP 进阶: | 知识点 | 档 | | --- | --- | | 组合优先于继承:什么时候继承是对的(is-a 且需要多态) | A | | super() 与 MRO(Cls.__mro__)、协作式 __init__ | A / B | | abc.ABC + @abstractmethod:强制子类实现;实例化抽象类报错 | A | | typing.Protocol:结构化子类型——不继承也算实现;@runtime_checkableisinstance | A | | @classmethod 备用构造器(from_dict)、@staticmethod@property + setter、类属性 vs 实例属性 | A | | __init_subclass__ 自动注册子类、__class_getitem____slots__ | B | | 描述符、元类、__new__ | C |

练习 w1_09_oop_protocols.pyclass Shape(ABC) + Circle/Rectarea/perimeter);class Drawable(Protocol) 与两个互不继承但都实现 draw() -> str 的类,render_all(items: Iterable[Drawable])class Repository(Protocol) 定义 add/get/list + InMemoryRepository 实现(下周 SQLite 版会再实现一次);class Temperature@property 做校验;class Plugin__init_subclass__ 自动注册到 Plugin.registryVehicle → Car/Truck 协作式 super().__init__(**kwargs)

下午前 2 块 — M5 异常进阶与 3.12–3.14 语法: | 知识点 | 档 | | --- | --- | | 包级异常层次:class AppError(Exception)NotFoundError/ValidationError;调用方只捕基类 | A | | raise NewError(...) from e(保留原因)vs from Nonee.__cause__e.add_note()(3.11) | A | | try/except/else/finally 的精确语义;finally 里不要 return(3.14 警告) | A | | ExceptionGroupexcept*(3.11)——并发任务多个失败(下周 TaskGroup 会遇到) | B | | warnings.warn(..., DeprecationWarning)warnings.deprecated(3.13);assert 不能用于校验(-O 会去掉) | B | | match 进阶:类模式 case Point(x=0)、守卫 if|、捕获与通配 | B | | PEP 695:class Stack[T]def first[T](xs: list[T]) -> Ttype Pair = tuple[int, int] | B | | t-strings(3.14):t"..." 得到 Template,可在拼接前处理各部分(安全拼 SQL/HTML/日志) | C | | 位置仅参数 def f(a, /, b, *, c) | B |

练习 w1_10_exceptions_syntax.py:设计 KBError → NoteNotFound/InvalidNote 层次;load_note(store, id)KeyError 转成 NoteNotFound ... from evalidate_many(items) -> list 收集全部错误后 raise ExceptionGrouphandle(cmd: Command)match 类模式分派;class Stack[T](PEP 695)带 push/pop/peekfirst[T]render(t: Template) -> str(t-string:把插值转义 <>)。

下午第 3 块 + 晚上 1 块 — 小项目:文本处理管线 exercises/week1/project_pipeline/

目标:一个命令行工具 uv run python exercises/week1/project_pipeline/pipeline.py <目录>,用生成器管线统计一个目录下全部 .md 文件的词频、标题数、代码块数,并输出报表。结构要求: - Step(Protocol)__call__(self, items: Iterable[Record]) -> Iterator[Record];每个步骤是生成器函数,用 @register_step("name") 注册到 Registry。 - 至少 5 个步骤:read_filesPath.rglob,生成器)、strip_code_blockssplit_wordsnormalizecount(返回 Stats dataclass:frozen=True)。 - Pipeline 类:__init__(steps: list[str]) 按名字从注册表取步骤,run(source) 依次串联(functools.reduce 或循环);__len____iter__ 遍历步骤名、__repr__。 - @timer 装饰 runTimer 上下文管理器包住整次运行;日志用 print(下周换 logging)。 - 异常:文件读取失败 → 自定义 PipelineError ... from e,跳过并记录,不中断。 - 测试 test_pipeline.py(已给 8 个):用 tmp_path 造 3 个 md 文件;测试每个步骤的惰性(传入生成器不预先耗尽)与结果。 - 验收:测试全绿、ruff check 零警告、对本仓库 research/ 目录运行能输出词频前 20。

晚上第 2 块 — 周测 + AI 评审exercises/week1/quiz_week1.md 20 题(闭卷 40 分钟,答案在 exercises/solutions/week1/quiz_week1_answers.md);把 pipeline.pyai-guide.md 模板 4 评审,只改前 3 条。写周日志与下周预览(读 research/08 第一节 10 分钟)。


本周阅读(详见 research/07research/13

模块 中文主线 英文/官方
M1 《流畅的 Python(第 2 版)》第 17 章 迭代器、生成器;廖雪峰《迭代器》《生成器》 Python 官方 Functional HOWTO;itertools 文档
M2 《流畅的 Python》第 7、9 章 Real Python "Primer on Python Decorators";functoolsinspect 文档
M3 《流畅的 Python》第 18 章 contextlib 文档
M4 《流畅的 Python》第 1、11、12、13 章;《Python 工匠》面向对象章节 官方 Data Model;dataclassesenum 文档
M5 官方教程第 8 章;What's New 3.11–3.14 PEP 654、PEP 695、PEP 750

本周陷阱清单(周测必考)

生成器只能遍历一次|groupby 前忘排序|装饰器忘 wraps|闭包晚绑定|nonlocal 忘写导致 UnboundLocalError__eq__ 不配 __hash__|可变类属性被所有实例共享|@dataclass 可变默认值|frozen 后在 __post_init__ 里赋值要用 object.__setattr__super().__init__() 漏传参数|except Exception: pass 吞错|raise 丢失原因(不用 from)|finallyreturnProtocol 类不该被实例化|matchcase x: 是捕获不是比较