学习笔记——测试进阶之路 WorkBuddy 带我学: Python 高级知识点完全指南

大海 · 2026年07月20日 · 637 次阅读

网页平台:https://pythonplus.netlify.app/

网页截图

Python 高级知识点完全指南

本文档是一份系统、深入的 Python 高级特性教程。每个主题都按"原理 → 示例 → 底层机制 → 常见陷阱 → 实践建议"展开,而非罗列代码片段。配套 9 幅图示,覆盖对象模型、函数式编程、元编程、执行模型、并发与性能。

阅读前提:熟悉 Python 基础语法(def/class/for/try、列表/字典、模块导入)、理解"一切皆对象"的直觉。本文不重复基础语法。

目录

  1. 知识版图总览
  2. 魔术方法(Dunder Methods)全览
  3. 描述符协议
  4. __slots__ 内存优化
  5. 闭包与装饰器
  6. functools 工具集
  7. 迭代器协议
  8. itertools 与函数式工具
  9. 上下文管理器
  10. 生成器与协程
  11. asyncio 异步编程
  12. super() 深度解析
  13. 元类(metaclass)
  14. 抽象基类 (ABC)
  15. collections 常用容器
  16. dataclasses
  17. 类型注解 (typing)
  18. 并发模型对比
  19. 内存管理与 GIL
  20. 性能分析与优化
  21. 学习路线图

1. 知识版图总览

Python 的"高级特性"看似零散,实则可以用两条主线串起来:

  • 对象模型(Object Model):Python 中"一切皆对象"。类本身是对象(type 的实例),方法其实是描述符,属性访问有固定的查找链。理解了对象模型,你就理解了装饰器、描述符、元类、魔术方法为什么那样工作。
  • 执行模型(Execution Model):函数调用如何压栈、yield 如何让函数挂起、事件循环如何调度协程、GIL 如何影响多线程。理解了执行模型,你就理解了生成器、asyncio、并发与性能调优。

下面这张图把全部主题归入 6 个分区,建议按"对象模型 → 函数式 → 执行模型 → 元编程 → 数据结构 → 性能"的顺序阅读。

本文的定位:不是语法手册,而是"为什么这样设计"与"什么时候该用"。所有代码均可直接运行(Python 3.10+ 验证)。遇到不确定的行为,优先用 python -iipython 亲手试。


2. 魔术方法(Dunder Methods)全览

核心概念:双下划线方法(dunder = double underscore)是 Python 对象模型的基础。当你写 a + blen(x)for v in objprint(obj) 时,解释器实际调用的是 obj.__add__(b)x.__len__()obj.__iter__()obj.__str__()。重载这些方法,就能让你的类"像内置类型一样自然地工作"。

2.1 构造与析构:__new__ / __init__ / __del__

这是最容易被误解的一组。__new____init__ 职责完全不同:

方法 调用时机 职责 返回值
__new__(cls, ...) 最先调用 分配对象内存(创建实例) 必须返回新实例
__init__(self, ...) __new__ 之后 初始化实例属性 必须返回 None(返回其他值会报错)
__del__(self) 对象被 GC 回收前 清理资源 无(返回值被忽略)
class Foo:
    def __new__(cls, *args, **kwargs):
        # 真正的"构造函数":负责把内存造出来
        print("__new__ 分配实例")
        instance = super().__new__(cls)
        return instance           # 关键:必须返回实例,否则 __init__ 不会被调用

    def __init__(self, value):
        print("__init__ 初始化")
        self.value = value        # 只赋值,不要 return

    def __del__(self):
        print("__del__ 析构")      # 不可靠:调用时机不确定

f = Foo(10)
# 输出: __new__ 分配实例
#       __init__ 初始化

为什么需要 __new__ 因为 __new__静态方法(第一个参数是类,不是实例),它可以在返回实例前做决定,甚至返回一个不同类的实例。典型应用场景:

# 不可变类型(int/str/tuple)和需要"缓存/池化"的对象必须重写 __new__
class PositiveInt(int):
    def __new__(cls, value):
        # int 是不可变的,必须在创建时修正值,__init__ 无法修改已创建好的 int
        return super().__new__(cls, abs(value))

PositiveInt(-5)   # 5 —— 负数被转成绝对值

__del__ 为什么不可靠? 它的调用时机由垃圾回收决定,不保证"出了作用域立即执行";循环引用还可能延迟它;更危险的是,__del__ 中抛出异常会被忽略。with 语句和上下文管理器才是资源清理的正确手段(见第 9 节)。

2.2 字符串表示:__repr__ / __str__ / __format__

  • __repr__:给开发者/调试器看,目标是"尽可能无歧义、最好能 eval 还原对象"。约定:repr(x) == "ClassName(args)"
  • __str__:给最终用户看,目标是"可读"。print(x)str(x) 优先用 __str__,没有才退回 __repr__
  • __format__:支持 format(x, spec) 与 f-string 的 :spec 部分。
class Point:
    def __init__(self, x, y):
        self.x, self.y = x, y

    def __repr__(self):
        # 首选:能直接看到如何重建这个对象
        return f"Point(x={self.x}, y={self.y})"

    def __str__(self):
        # 给用户:简洁可读
        return f"({self.x}, {self.y})"

    def __format__(self, spec):
        # 支持 f"{p:.2f}" 之类
        if spec == ".2f":
            return f"({self.x:.2f}, {self.y:.2f})"
        return str(self)

p = Point(3.14159, 2.71828)
print(repr(p))        # Point(x=3.14159, y=2.71828)
print(str(p))         # (3.14159, 2.71828)
print(f"{p:.2f}")     # (3.14, 2.72)

经验法则:每个"数据类"都应实现 __repr__——调试时能立刻看清对象内容,省去大量 print(obj.__dict__)

2.3 运算符重载

重载运算符时,必须理解三组规则:反射运算符原地运算符、以及 __eq__/__hash__ 的契约

class Vector:
    def __init__(self, x, y):
        self.x, self.y = x, y

    def __add__(self, other):           # self + other
        return Vector(self.x + other.x, self.y + other.y)

    def __radd__(self, other):          # other + self(当 other 没有 __add__ 时反射调用)
        return Vector(other + self.x, other + self.y)

    def __iadd__(self, other):          # self += other(原地,应返回 self)
        self.x += other.x
        self.y += other.y
        return self

    def __eq__(self, other):            # ==
        return isinstance(other, Vector) and self.x == other.x and self.y == other.y

    def __hash__(self):                 # 与 __eq__ 配套
        return hash((self.x, self.y))

    def __lt__(self, other):            # <(排序用)
        return (self.x, self.y) < (other.x, other.y)

三大规则:

  1. 反射运算符(__radd____rmul__ 等):当 a + ba.__add__(b) 返回 NotImplemented 时,Python 会尝试 b.__radd__(a)。这是让 1 + Vector(...)(数字在左)也能工作的关键。
  2. 原地运算符(__iadd__ 等)a += b 优先调 __iadd__,返回 self 实现真正的原地修改(可变对象省内存)。不可变对象(如自定义元组)应不实现 __iadd__,退回 __add__
  3. __eq____hash__ 的契约(极重要):如果两个对象 __eq__ 相等,它们的 __hash__ 必须相等。一旦你重写了 __eq__,Python 会自动把类变成不可哈希(删掉 __hash__);想让它可哈希,必须显式重写 __hash__。反之,若对象可变且参与哈希集合,应在 __hash____eq__ 上保持一致性。

陷阱:只重写 __eq__ 不写 __hash__,对象就不能放进 set/dict 的键——常导致 TypeError: unhashable type。列表 (list)、字典 (dict) 等可变内置类型正是因此不可哈希。

2.4 容器模拟

让自定义类表现得像序列/映射:

class MyList:
    def __init__(self, data):
        self._data = list(data)

    def __len__(self):                # len(obj)
        return len(self._data)

    def __getitem__(self, idx):       # obj[idx],支持切片(idx 可能是 slice)
        return self._data[idx]

    def __setitem__(self, idx, val):  # obj[idx] = val
        self._data[idx] = val

    def __delitem__(self, idx):       # del obj[idx]
        del self._data[idx]

    def __contains__(self, val):      # val in obj
        return val in self._data

    def __iter__(self):               # for x in obj
        return iter(self._data)

    def __reversed__(self):           # reversed(obj)
        return reversed(self._data)

要点

  • __getitem__ 收到的 idx 可能是整数,也可能是 slice 对象(如 obj[1:3])。若要支持切片,需自行判断 isinstance(idx, slice)
  • 实现了 __getitem__ + __len__ 的对象就满足 collections.abc.Sequence 的鸭子类型,可当作序列使用。
  • 想让类成为"映射",还需实现 __setitem__/__delitem__/keys/values/items 等,或用 collections.UserDict 包装(见第 15 节)。

2.5 可调用与属性访问

class CallableObj:
    def __call__(self, *args, **kwargs):
        print("被当作函数调用:", args)
        return sum(args)

f = CallableObj()
f(1, 2, 3)   # 被当作函数调用: (1, 2, 3) → 6

__call__实例像函数一样可调用。常见于需要"持有状态的函数":缓存器、配置化处理器、策略对象。

属性访问拦截是高级但危险的领域,务必分清三者:

方法 触发时机 风险
__getattr__(self, name) 常规属性查找失败后(兜底) 安全,常用于代理/惰性加载
__getattribute__(self, name) 每次属性访问都触发 极易无限递归,必须 super().__getattribute__
__setattr__(self, name, val) 每次属性赋值 同样易递归,必须 super().__setattr__
class Proxy:
    def __init__(self, target):
        self._target = target

    def __getattr__(self, name):
        # 只有找不到 name 时才走到这里(_target 自身属性不会触发)
        return getattr(self._target, name)

class Dangerous:
    def __getattribute__(self, name):
        # 每次访问属性都触发,包括读取 self._x 也会再触发 → 无限递归
        # 正确写法:
        return super().__getattribute__(name)

    def __setattr__(self, name, value):
        # 错误写法:self._data = value 又会触发 __setattr__ → 递归
        # 正确写法:
        super().__setattr__(name, value)

属性访问的黄金法则__getattr__ 是"安全兜底",优先用它;__getattribute__/__setattr__ 必须用 super() 委托,否则直接栈溢出。绝大多数场景不需要它们——描述符(第 3 节)是更可控的属性拦截方案。

补充:__missing__(字典专属)dict 子类实现 __missing__(key),可在 d[key] 缺失时返回默认值而不抛 KeyError。这是 collections.defaultdict 的底层机制。


3. 描述符协议

核心概念:描述符是实现了 __get____set____delete__至少一个的对象。它表面上不起眼,却是 Python 属性系统的底层引擎——你每天用的 propertyclassmethodstaticmethod,以及 Django、SQLAlchemy、Pydantic 里的字段,全都是描述符。

3.1 描述符协议与 data/non-data 之分

描述符按"是否定义了 __set__"分为两类,这决定了它的优先级:

  • data descriptor(数据描述符):定义了 __set__(可能还有 __delete__)。优先级最高,连实例字典都压不过它。
  • non-data descriptor(非数据描述符):只有 __get__(如普通函数/方法)。优先级低于实例字典。
class Validated:
    def __init__(self, validator):
        self.validator = validator

    def __set_name__(self, owner, name):
        # Python 3.6+:描述符被赋给类属性时,自动得知自己的名字
        self.name = f"_{name}"

    def __get__(self, inst, owner):
        # inst 为实例;通过类访问时 inst 为 None
        if inst is None:
            return self
        return getattr(inst, self.name, None)

    def __set__(self, inst, value):
        if not self.validator(value):
            raise ValueError(f"校验失败: {self.name}")
        setattr(inst, self.name, value)

class User:
    name = Validated(str.isalpha)                 # 仅字母
    age  = Validated(lambda x: isinstance(x, int) and 0 < x < 150)

u = User()
u.name = "Alice"
u.age = 30
# u.age = -1      # ValueError: 校验失败
# u.name = "Bob1" # ValueError: 校验失败
  • __get__(self, inst, owner)owner 是拥有该描述符的类。inst is None 表示通过访问(如 User.name),此时通常返回描述符本身。
  • __set_name__ 让你不必手写字符串属性名,框架里常用来自动登记字段。

3.2 属性解析顺序(为什么"实例属性可以遮蔽方法")

当你写 obj.x,Python 按以下固定顺序查找:

  1. 如果 xdata descriptor(有 __set__)→ 调用它的 __get__。优先级最高。
  2. 否则,查 实例字典 obj.__dict__['x']
  3. 否则,查 类字典 type(obj).__dict__['x'],若找到的是 non-data descriptor(如函数)→ 调用 __get__(这把函数变成"绑定方法")。
  4. 否则,查父类的字典(按 MRO)。
  5. 否则,调用类的 __getattr__(若有)。再找不到 → AttributeError

这就是关键洞察:普通方法是 non-data descriptor。因此当你写 obj.foo = 123,实例字典里有了 foo,它就遮蔽了类上的 foo 方法(因为步骤 2 在步骤 3 之前)。而 property 是 data descriptor,所以即使你给实例赋值 obj.x = ...,也会被 property 的 __set__ 拦截——实例字典里根本不会出现 x

property 本质property(fget, fset, fdel, doc) 就是一个 data descriptor 工厂。@property 是语法糖,等价于创建一个带 __get__/__set__ 的描述符实例并赋给类属性。

3.3 描述符的典型实战

缓存描述符(lazy property):首次访问计算,之后缓存到实例字典。

class cached_property:
    def __init__(self, func):
        self.func = func
        self.name = func.__name__

    def __set_name__(self, owner, name):
        self.name = name

    def __get__(self, inst, owner):
        if inst is None:
            return self
        # 算一次后直接写进实例字典:下次查找步骤 2 命中,不再走描述符
        value = self.func(inst)
        inst.__dict__[self.name] = value
        return value

class Report:
    def __init__(self, data):
        self.data = data

    @cached_property
    def summary(self):
        print("计算一次...")
        return sum(self.data)

r = Report([1,2,3])
r.summary   # 计算一次... → 6
r.summary   # 直接读缓存 → 6

提示functools.cached_property 内置了这个模式(Python 3.8+),生产代码直接用它即可。


4. __slots__ 内存优化

原理:普通对象每个实例都带一个 __dict__(字典),用来存放任意动态属性。字典本身有开销(哈希表、指针数组),且无法高效紧凑排列。声明 __slots__ 后,Python 不再为实例创建 __dict__,而是把列出的属性变成固定大小的"槽位"(类似 C 结构体),直接存在对象内存里。

class Point:
    __slots__ = ('x', 'y')          # 只允许这两个属性
    def __init__(self, x, y):
        self.x, self.y = x, y

p = Point(1, 2)
# p.z = 3   # AttributeError: 'Point' object has no attribute 'z'

4.1 内存与速度

100 万个实例时,差距非常显著:

import sys, time, tracemalloc

class WithDict:
    def __init__(self, x, y): self.x, self.y = x, y

class WithSlots:
    __slots__ = ('x', 'y')
    def __init__(self, x, y): self.x, self.y = x, y

def bench(cls, n=1_000_000):
    tracemalloc.start()
    objs = [cls(i, i) for i in range(n)]
    cur, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return peak / 1024 / 1024

print(f"普通类: {bench(WithDict):.1f} MB")     # ~150 MB
print(f"slots:  {bench(WithSlots):.1f} MB")    # ~55 MB(约省 60%)

4.2 使用代价与陷阱

__slots__ 不是免费的,务必注意:

收益 代价
省内存(大量实例时显著) 不能动态添加未在 __slots__ 中声明的属性
属性访问略快(少一层字典查找) 多继承时,只有所有父类都定义 __slots__ 才有效
可阻止意外属性名拼写错误 默认没有 __dict__,需要弱引用支持时还要加 '__weakref__'
继承链复杂时易踩坑
class Node:
    __slots__ = ('value', '__weakref__')   # 若要用 weakref.ref,需显式加入 __weakref__
    def __init__(self, v): self.value = v

决策准则:仅当实例数量达到十万级以上属性集合固定时才值得用(如 ORM 行对象、海量数据点)。日常业务对象、配置类、DTO 用 @dataclass 更合适——可读性与灵活性远比那点内存重要。__slots__ = () 还可让类本身不可实例化出有 __dict__ 的对象。


5. 闭包与装饰器

5.1 闭包(Closure)的本质

概念:内层函数捕获了外层作用域的变量(自由变量),即使外层函数已经返回,这些变量依然存活——因为 Python 用"cell 对象"持有它们,形成闭包。

def make_multiplier(factor):
    def multiply(x):
        return x * factor     # factor 是自由变量,被闭包捕获
    return multiply

double = make_multiplier(2)
triple = make_multiplier(3)
print(double(5), triple(5))   # 10  15
# make_multiplier 已返回,但 factor=2 / 3 仍被 double/triple 持有

验证闭包捕获的变量:

print(double.__closure__[0].cell_contents)   # 2

关键点:multiply 没有自己的 factor,它引用的是外层作用域的 factor。每次调用 make_multiplier 都会创建一个新的 cell。注意循环 + 闭包的经典陷阱(见下)。

5.2 装饰器(Decorator)= "接收函数、返回函数"的高阶函数

装饰器本质就是语法糖:@deco 等价于 func = deco(func)。它把闭包用于"在不修改原函数代码的前提下,增强其行为"。

import functools, time

def timer(func):
    @functools.wraps(func)           # 关键:保留原函数的 __name__/__doc__/__module__
    def wrapper(*args, **kwargs):
        t = time.perf_counter()
        result = func(*args, **kwargs)
        print(f"{func.__name__} 耗时 {time.perf_counter()-t:.4f}s")
        return result                # 必须原样返回原函数的返回值
    return wrapper

@timer
def add(a, b):
    """两数相加"""
    return a + b
# 等价于: add = timer(add)

add(3, 4)          # 打印耗时并返回 7
print(add.__name__)  # "add"(因为 @wraps);否则会是 "wrapper"

为什么 @functools.wraps 不可省略? 没有它,被装饰函数的元信息(__name____doc____module__、参数签名)会变成 wrapper 的,导致:日志显示错误函数名、inspect.signature 失效、pickle 失败、基于 __name__ 的路由/注册出错。

5.3 带参装饰器(三层嵌套)

当装饰器本身要接收参数时,就变成三层结构:最外层接收参数,返回"真正的装饰器",再由它包裹函数。

import functools, time

def retry(max_attempts=3, delay=1):
    def decorator(func):                     # 第二层:真正的装饰器
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception:
                    if i == max_attempts - 1:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator                          # 第一层返回装饰器

@retry(max_attempts=5, delay=0.5)
def fetch_data():
    ...

难点@retry@retry(...) 写法不同。@retry 时 Python 把被装饰函数直接传给 retry@retry(5) 时先调用 retry(5) 得到装饰器再包裹。要同时支持两种写法,可在第一层判断第一个参数是否是可调用的函数(略复杂,生产建议分开)。

5.4 类装饰器(持有状态)

当装饰器需要持有状态(如计数、缓存字典),用类更自然——类实例本身就是状态容器。

import functools

class CountCalls:
    def __init__(self, func):
        functools.update_wrapper(self, func)   # 等价于 wraps
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        return self.func(*args, **kwargs)

@CountCalls
def greet(name):
    return f"hi {name}"

greet("A"); greet("B")
print(greet.count)   # 2

5.5 装饰器叠加(执行顺序)

多个装饰器从下往上包裹(离函数最近的先应用):

@a
@b
@c
def f(): ...
# 等价于: f = a(b(c(f)))   # c 先包,a 最后包;调用时 a→b→c→原函数的层层进入

5.6 装饰器本质图

实战清单:计时、日志、重试、缓存 (lru_cache)、权限校验、注册路由、单例、弃用警告 (warnings) 都是装饰器的典型用途。记住铁律——装饰器必须返回可被调用且能透传返回值/异常的对象,并保持 @wraps


6. functools 工具集

functools 是函数式编程的工具箱。以下几项是高频必备。

6.1 partial — 偏函数(冻结部分参数)

from functools import partial

def power(base, exp):
    return base ** exp

square = partial(power, exp=2)    # 固定 exp=2
cube   = partial(power, exp=3)
print(square(5), cube(4))         # 25  64

# 实战:给回调预置参数(GUI/异步里非常常见)
on_click = partial(handle_click, button_id=7)

partial 返回的对象会保留 __name__/__doc__,但 inspect.signature 会反映剩余参数。注意 partial 是浅冻结:若冻结的是可变对象,调用方仍可能修改它。

6.2 lru_cache — 记忆化缓存

把函数结果按参数缓存,相同输入直接返回,避免重复计算。

from functools import lru_cache

@lru_cache(maxsize=128)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

print(fib(100))                 # 毫秒级(否则指数爆炸)
print(fib.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=128, currsize=101)

关键细节

  • 缓存键是参数的哈希:被装饰函数的所有参数必须可哈希(__hash__)。传列表会报错——先转元组。
  • maxsize=None 表示无限缓存;maxsize=2**n 性能最佳(用环形链表)。
  • typed=True 时,f(3)f(3.0) 视为不同键。
  • fib.cache_clear() 清空缓存。
  • 适用于纯函数(相同输入必得相同输出,无副作用)。

6.3 singledispatch — 单分派泛型函数

第一个参数类型自动选择实现,是"重载"的优雅替代(Java/C++ 才有真正重载,Python 用这个)。

from functools import singledispatch

@singledispatch
def process(data):
    raise NotImplementedError(f"不支持类型: {type(data)}")

@process.register
def _(data: int):
    return data * 2

@process.register
def _(data: str):
    return data.upper()

@process.register(list)
def _(data):                       # 也可写成 register(list)
    return [process(x) for x in data]

print(process(42))          # 84
print(process("hello"))     # HELLO
print(process([1, "a", 3])) # [2, 'A', 6]

Python 3.11+ 还有 singledispatchmethod:用于类方法的分派。

6.4 reduce — 累积归约

from functools import reduce

product = reduce(lambda a, b: a * b, [1, 2, 3, 4])        # 24
max_val = reduce(lambda a, b: a if a > b else b, [3, 7, 2, 9])  # 9
# 可给初始值(避免空序列报错):reduce(fn, seq, init)

多数情况下 functools.reduce 不如显式 forsum/max 可读。仅当"把序列折叠成一个值"且逻辑紧凑时使用。

6.5 wraps / update_wrapper

已见 5.2。补充:update_wrapper 是底层 API,wraps 是它的装饰器封装。它复制 __module____name____qualname____doc____dict__ 等,并设 __wrapped__ 指向原函数(供 inspect 穿透)。


7. 迭代器协议

核心概念:Python 的 for 循环、解包、* 展开、in 判断,底层都依赖迭代器协议。理解它,就理解了"惰性计算"和"流式处理"的基础。

  • 可迭代对象(Iterable):实现 __iter__() 返回迭代器的对象(如 liststrdictset、生成器)。
  • 迭代器(Iterator):同时实现 __iter__()(返回自身)和 __next__()(返回下一个元素;耗尽时抛 StopIteration)的对象。
class Countdown:
    def __init__(self, start):
        self.start = start

    def __iter__(self):        # 返回迭代器(这里就是自己)
        return self

    def __next__(self):        # 每次调用推进一次
        if self.start <= 0:
            raise StopIteration
        self.start -= 1
        return self.start + 1

for n in Countdown(3):
    print(n)                   # 3 2 1

# 底层完全等价:
it = iter(Countdown(3))        # 调用 __iter__
while True:
    try:
        print(next(it))        # 调用 __next__
    except StopIteration:
        break

迭代器 vs 可迭代对象的关系

关键区别:迭代器是"一次性"的——走完(StopIteration)后再 next() 仍抛异常,且不会自动重置。列表是可迭代但不是迭代器(每次 iter(lst) 都返回全新的迭代器);而生成器本身就是迭代器。

nums = [1, 2, 3]
it1 = iter(nums); it2 = iter(nums)
next(it1); next(it1)           # 1, 2
next(it2)                      # 1 —— 独立的迭代器,互不影响

StopIteration 的现代陷阱(PEP 479):在生成器里,若正常 return 会以 StopIteration 结束;但如果在生成器内部意外抛出 StopIteration,Python 3.7+ 会把它转换为 RuntimeError,防止它错误地终止外层 for。这是历史包袱的修复,理解即可。

实战价值:自定义迭代器让你能处理无限流超大数据集而不占内存——每次只产生一个元素。这也是 itertools生成器 的基石。


8. itertools 与函数式工具

itertools 提供一组惰性、内存友好的迭代器,是高效数据处理利器。下面逐一给出用法与输出。

import itertools

# --- 无限迭代器(务必配合 islice / 条件终止,否则死循环)---
itertools.count(10, 2)            # 10, 12, 14, 16, ...(从 10 起步长 2)
itertools.cycle("AB")             # A, B, A, B, A, B, ...(无限循环)
itertools.repeat("x", 3)          # x, x, x(有限次,可省略次数成无限)

# --- 组合迭代器 ---
itertools.product("AB", "12")     # 笛卡儿积
# ('A','1'), ('A','2'), ('B','1'), ('B','2')
itertools.permutations("AB", 2)   # 排列(有序):('A','B'), ('B','A')
itertools.combinations("ABC", 2)  # 组合(无序):('A','B'), ('A','C'), ('B','C')
itertools.combinations_with_replacement("AB", 2)  # 允许重复:('A','A'),('A','B'),('B','B')

# --- 终止迭代器 ---
list(itertools.accumulate([1, 2, 3, 4]))       # 前缀和: [1, 3, 6, 10]
list(itertools.chain([1, 2], "ab"))            # 拼接: [1, 2, 'a', 'b']
list(itertools.chain.from_iterable([[1,2],[3,4]]))  # 展平: [1, 2, 3, 4]
list(itertools.groupby("aaabbc"))              # 按键分组(注意:键必须已排序!)
# [('a', <g>), ('b', <g>), ('c', <g>)]  每个值是迭代器
list(itertools.islice(range(100), 5, 15))      # 切片(惰性): [5..14]
list(itertools.zip_longest("AB", "123"))       # 不等长补 None: [('A','1'),('B','2'),(None,'3')]
list(itertools.pairwise([1, 2, 3, 4]))         # 3.10+ 滑动窗口: [(1,2),(2,3),(3,4)]

常见实用模式

# 1) 分组前必须先排序(groupby 只连续归组)
data = sorted([("a",1),("b",2),("a",3)])
for key, grp in itertools.groupby(data, key=lambda x: x[0]):
    print(key, list(grp))
# a [('a',1),('a',3)]
# b [('b',2)]

# 2) 滑窗统计(如滚动平均)
def windowed(seq, n):
    iters = itertools.tee(seq, n)              # 复制 n 份迭代器
    for i, it in enumerate(iters):
        next(itertools.islice(it, i, i), None) # 各自偏移 i
    return zip(*iters)
list(windowed([1,2,3,4,5], 3))   # [(1,2,3),(2,3,4),(3,4,5)]

# 3) 生成器表达式(最 Pythonic 的惰性映射+过滤)
squares = (x*x for x in range(10) if x % 2 == 0)   # 不立即计算,省内存

tee 会把迭代器"快照"成多份,但会缓冲已消费的元素——若只需顺序访问,优先考虑 islice/手动游标,避免 tee 占用过多内存。


9. 上下文管理器

核心概念:上下文管理器让资源"自动清理"——无论代码块正常结束还是抛异常,__exit__(或 yield 之后)都会运行。这是 try/finally 的优雅封装,用于文件、锁、连接、计时、临时状态等。

9.1 类实现:__enter__ / __exit__

import time

class Timer:
    def __enter__(self):
        self.t = time.perf_counter()
        return self                 # with ... as <val> 接收的就是它的返回值

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"耗时 {time.perf_counter()-self.t:.4f}s")
        return False                # False=不吞异常(向外传播);True=吞掉异常

with Timer() as t:
    time.sleep(0.5)
# 输出: 耗时 0.50xx s
# sleep 中若抛异常,__exit__ 仍会被调用,且异常继续向外抛出(因返回 False)

__exit__ 的三个参数在无异常时全为 None;有异常时分别为异常类型、实例、回溯对象。

9.2 contextmanager — 生成器方式(更简洁)

@contextmanager 把"含 yield 的生成器"变成上下文管理器:yield 之前是进入逻辑,yield 之后是退出逻辑。

from contextlib import contextmanager
import time, os

@contextmanager
def timer():
    t = time.perf_counter()
    yield                          # 这里的 yield 值 = as 接收的值(此处为 None)
    print(f"耗时 {time.perf_counter()-t:.4f}s")

@contextmanager
def temp_env(key, value):          # 临时修改环境变量,退出还原
    old = os.environ.get(key)
    os.environ[key] = value
    try:
        yield
    finally:                       # 即使块内异常,也确保还原
        if old is None:
            os.environ.pop(key, None)
        else:
            os.environ[key] = old

with temp_env("MODE", "prod"):
    print(os.environ["MODE"])      # prod
print(os.environ.get("MODE"))      # 恢复为 None

异常处理@contextmanager 包裹的函数里,若 yield 之后的代码抛异常(如 finally 中的清理也出错),该异常会传播;若块内抛异常,它会在 yield 处重新抛出,由 try/except 捕获或传播。务必用 try/finally 保证清理。

9.3 其他 contextlib 工具

from contextlib import suppress, closing, ExitStack, nullcontext, redirect_stdout
import os, io, sys

with suppress(FileNotFoundError):   # 优雅忽略指定异常(不必写 try/except/pass)
    os.remove("nonexistent.txt")

with closing(open("f.txt")) as f:   # 确保 .close() 被调用(对象本身不支持 with 时)
    data = f.read()

with ExitStack() as stack:          # 动态管理数量不定的多个上下文
    files = [stack.enter_context(open(f"file{i}.txt")) for i in range(3)]
    # 退出时全部自动关闭,按逆序

with redirect_stdout(io.StringIO()) as buf:  # 临时重定向 stdout(测试很有用)
    print("hello")
captured = buf.getvalue()           # "hello\n"

with nullcontext() as c:            # 占位上下文(根据条件决定是否套真实上下文)
    pass

上下文管理器执行流程

额外contextlib.asynccontextmanager 提供异步版本的 @contextmanager,配合 async with 使用(见第 11 节)。


10. 生成器与协程

10.1 生成器基础(惰性 + 挂起)

yield 让函数在产出值后挂起并保留局部状态,下次被唤醒时从暂停点继续。与一次性算出全部结果的列表相比,生成器按需产生、几乎零内存

# 生成器表达式(惰性)
g = (x*x for x in range(10) if x % 2 == 0)
print(next(g), next(g))     # 0 4

# yield from:委托给子生成器(自动转发 yield/值,并收集子生成器 return 的值)
def chain(*iters):
    for it in iters:
        yield from it        # 等价于 for x in it: yield x
print(list(chain([1, 2], (3, 4))))   # [1, 2, 3, 4]

10.2 协程:双向通信的生成器

生成器不仅能产出(yield 给调用者),还能通过 send() 接收值——这就成了协程(双向数据流)。throw() 注入异常,close() 终止。

def accumulator():
    total = 0
    while True:
        x = yield total           # 暂停,把 total 交出去,等待 send 注入到 x
        if x is None:
            break
        total += x

acc = accumulator()
next(acc)            # 预激活:必须先 next() 或 send(None) 跑到第一个 yield
print(acc.send(10))  # 10  (total=10)
print(acc.send(5))   # 15  (total=15)
acc.send(None)       # 结束(break → StopIteration)
# acc.close()        # 也可主动关闭,触发 GeneratorExit

生命周期状态机

状态 说明
CREATED 调用生成器函数,返回生成器对象;函数体尚未执行
RUNNING 调用 next()/send() 后进入,执行函数体直到遇到 yield
SUSPENDED yield 处挂起,把值交给调用者;等待下一次唤醒
CLOSED 函数执行完毕(return/StopIteration)、被 close() 或抛异常,生命周期结束

10.3 协程的实战价值与 yield from 异常传播

yield from 不仅是语法糖,它还会:把子生成器的 yield 直接透传给最外层、把 send 直接转发给子生成器、把子生成器的 return value 作为 yield from 表达式的值(PEP 380 的核心贡献)。

def sub():
    yield 1
    return "done"

def outer():
    result = yield from sub()     # result == "done"
    yield result

g = outer()
print(next(g))      # 1
print(next(g))      # "done"

10.4 生成器的最佳场景

  • 数据管道:逐行读大文件、逐条处理流数据(日志、网络包),内存恒定。
  • 无限序列:斐波那契、传感器采样,永不"全量计算"。
  • 协程流水线send/throw 构建生产者 - 消费者。
  • 惰性组合itertools + 生成器表达式替代多轮列表拷贝。

现代写法:需要双向协程时,优先用 async/await(见第 11 节)——它语义更清晰、配套设施更全。手写 yield 协程如今主要用于数据管道和惰性求值。


11. asyncio 异步编程

核心概念asyncio 提供单线程事件循环 + 协作式调度。协程在 await主动交还控制权,事件循环转去运行其他就绪任务;IO 完成后再唤醒它。没有抢占、没有线程切换开销——因此能以极低代价支撑海量并发 IO。

11.1 最小模型

import asyncio, time

async def fetch(i):
    await asyncio.sleep(1)        # 模拟 IO:此处让出控制权,不阻塞线程
    return i * 10

async def main():
    t = time.perf_counter()
    results = await asyncio.gather(fetch(1), fetch(2), fetch(3))
    print(results, f"用时 {time.perf_counter()-t:.2f}s")

asyncio.run(main())   # [10, 20, 30] 用时约 1.00s(而非 3s)

关键术语

概念 说明
coroutine async def 返回的对象本身不执行,必须被 await 或包成 Task
Task asyncio.create_task(coro) 把协程登记到事件循环,立即开始调度(并发的真正来源)
await 只对 awaitable 生效;阻塞当前协程直到完成,但不阻塞事件循环
Event Loop 调度中枢:维护就绪队列,IO 完成回调唤醒对应 Task

11.2 并发的正确与错误写法

async def fetch(i):
    await asyncio.sleep(1)
    return i

# ❌ 错误:顺序 await → 退化成串行(总耗时 3s)
async def wrong():
    r1 = await fetch(1)
    r2 = await fetch(2)
    r3 = await fetch(3)

# ✅ 正确:先把协程变成 Task 并发调度
async def right():
    t1 = asyncio.create_task(fetch(1))
    t2 = asyncio.create_task(fetch(2))
    t3 = asyncio.create_task(fetch(3))
    results = await asyncio.gather(t1, t2, t3)   # 总耗时约 1s

# gather 也支持直接传协程(内部会自动 create_task)
results = await asyncio.gather(fetch(1), fetch(2), fetch(3))

11.3 生产者 - 消费者(Queue)

import asyncio

async def worker(q, wid):
    while True:
        item = await q.get()
        if item is None:
            q.task_done()
            break
        await asyncio.sleep(0.1)         # 模拟处理
        print(f"worker {wid} 处理 {item}")
        q.task_done()

async def producer(q):
    for i in range(5):
        await asyncio.sleep(0.05)
        await q.put(i)
    for _ in range(3):                    # 发 3 个 None 让 3 个 worker 退出
        await q.put(None)

async def main():
    q = asyncio.Queue()
    workers = [asyncio.create_task(worker(q, i)) for i in range(3)]
    await asyncio.gather(producer(q), *workers)

asyncio.run(main())

11.4 同步阻塞函数的处理(最重要陷阱)

在协程里调用同步阻塞函数(如 requests.gettime.sleepopen 读大文件)会卡死整个事件循环——因为阻塞期间控制权无法让出,所有其他任务都被冻住。解决方案:

import asyncio, concurrent.futures, requests

async def main():
    loop = asyncio.get_running_loop()
    # 把阻塞函数丢到线程池执行,await 等待其完成后不阻塞循环
    resp = await loop.run_in_executor(
        None, requests.get, "https://example.com"
    )
    # 或者用 asyncio.to_thread(Python 3.9+,语义相同,更简洁)
    data = await asyncio.to_thread(blocking_io_func, arg)
  • IO 阻塞(网络/磁盘)→ 换异步库(aiohttphttpx.AsyncClient)或 run_in_executor/to_thread
  • CPU 密集 → 用 loop.run_in_executor 丢到 ProcessPoolExecutor,或用 asyncio 之外的 multiprocessing

11.5 调度循环图示

更多要点asyncio.wait 可设 FIRST_COMPLETED/FIRST_EXCEPTION 精细控制;asyncio.TimeoutErrorasyncio.wait_for(coro, timeout) 做超时;asyncio.shield 防止取消传播;asyncio.CancelledError 用于优雅取消;@asynccontextmanager 写异步上下文管理器;async for 遍历异步迭代器。取消时用 task.cancel() 并在协程内 try/finally 清理。


12. super() 深度解析

最大的误解super() 不是"父类的引用"。它返回的是一个代理对象,沿当前实例的 MRO(方法解析顺序)链,查找调用者的下一个类。这个区分在多继承时至关重要。

12.1 单继承 vs 多继承的真相

class A:
    def foo(self):
        print("A.foo")

class B(A):
    def foo(self):
        print("B.foo")
        super().foo()          # 沿 MRO 找 B 之后的下一个 foo(即 A.foo)

class C(A):
    def foo(self):
        print("C.foo")
        super().foo()

class D(B, C):
    def foo(self):
        print("D.foo")
        super().foo()

D().foo()
# 输出:
# D.foo
# B.foo
# C.foo
# A.foo

D 的 MRO 是 [D, B, C, A, object]D.foo 里的 super() 找的是 MRO 中 D 之后第一个实现了 foo 的类 = BB.foo 里的 super()B 之后 = CC.foo 里的 super()C 之后 = A注意 Bsuper() 跳到了"兄弟类" C——这正是协作式多继承(cooperative MRO)的精髓。

12.2 super() 的两种形式

super()                       # 零参数:Python 3 自动填入当前类和 self
super(B, self)                # 显式:沿 B 在 self 的 MRO 之后的类查找
super(__class__, self)        # 函数内部用(避免硬编码类名)

零参数 super() 实际上是语法糖:编译器把当前 __class__ 和第一个参数(self/cls)注入进去。

12.3 协作式多继承的铁律

要让 super() 链不断裂,所有协作者必须:

  1. 都使用 super() 调用链上的下一个方法;
  2. 方法签名一致(通常参数都带 **kwargs 透传);
  3. 链的终点(通常是 object)能处理剩余参数。
class Base:
    def __init__(self, x, **kwargs):
        super().__init__(**kwargs)    # 把剩余 kwargs 传给下一个
        self.x = x

class Mixin:
    def __init__(self, y, **kwargs):
        super().__init__(**kwargs)
        self.y = y

class Derived(Base, Mixin):
    def __init__(self, z, **kwargs):
        super().__init__(**kwargs)    # 沿 MRO: Derived→Base→Mixin→object
        self.z = z

d = Derived(x=1, y=2, z=3)            # 三个 __init__ 都被正确调用

经验:不要滥用手写多继承 + super() 协作链——复杂度高且易错。能用组合(has-a)就用组合。super() 的常规正确用法是单继承里调用父类方法(最常见、最安全)。


13. 元类(metaclass)

核心概念:元类是"创建类的类"。在 Python 中,class 语句并非"声明",而是执行:解释器收集类体命名空间,再调用 type(name, bases, ns)(或指定的元类)构造出类对象。所以"类"也是对象——type 的实例。元类就是 type 的子类,拦截"类被创建"的瞬间。

13.1 type 的三参数形式

# 动态创建类的等价写法:
def say_hi(self):
    return f"hi from {self.name}"

Attrs = type("Attrs", (object,), {"name": "x", "say_hi": say_hi})
obj = Attrs()
print(obj.say_hi())   # hi from x

这等价于 class Attrs: name="x"; def say_hi(self): ...

13.2 自定义元类

class LogMeta(type):
    def __new__(mcs, name, bases, ns):
        # __new__ 负责"造出类对象"
        print(f"正在创建类: {name}, 基类: {bases}")
        for k, v in ns.items():
            if callable(v) and not k.startswith("_"):
                ns[k] = debug_wrapper(v)      # 在类创建期改写方法
        return super().__new__(mcs, name, bases, ns)

    def __init__(cls, name, bases, ns):
        # __init__ 负责"初始化已造好的类对象"
        super().__init__(name, bases, ns)
        cls.__author__ = "meta"

class Base(metaclass=LogMeta):
    def foo(self): ...
    def bar(self): ...
# 打印: 正在创建类: Base, 基类: (<class 'object'>,)
# Base.foo / Base.bar 被自动包上日志;其所有子类也会经 LogMeta 创建

__new__ vs __init__(元类)__new__ 返回新类对象;__init__ 在类对象造好后做初始化。多数"改造类"的逻辑放 __new____prepare__(可选)还能定制类体命名空间的字典类型(如自动排序)。

13.3 实战:自动注册子类(插件系统)

class PluginMeta(type):
    registry = {}

    def __new__(mcs, name, bases, ns):
        cls = super().__new__(mcs, name, bases, ns)
        if bases:                       # 跳过基类自身
            PluginMeta.registry[name.lower()] = cls
        return cls

class Plugin(metaclass=PluginMeta):
    pass

class JsonExporter(Plugin): ...
class CsvExporter(Plugin): ...

print(PluginMeta.registry)
# {'jsonexporter': <class ...>, 'csvexporter': <class ...>}

13.4 更轻量的替代:优先用它们而非元类

90% 的场景不需要元类。现代替代方案:

# 1) __init_subclass__(Python 3.6+):子类创建时回调,无需元类
class Plugin:
    registry = {}
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        Plugin.registry[cls.__name__.lower()] = cls

# 2) __set_name__:描述符自动获知属性名(见第 3 节)

13.5 类也是对象(对象链图示)

元类 vs 类装饰器:元类影响该类及其所有子类(在继承链创建时生效);类装饰器只作用在当前类上。决策顺序:能用函数/组合解决就别用装饰器;能用 __init_subclass__ 就别用元类;能用元类就别用更复杂的反射。元类适合框架级"批量改造类"(ORM、序列化器、API 路由自动注册),业务代码几乎用不到。


14. 抽象基类 (ABC)

核心概念:抽象基类定义接口契约——声明"必须实现哪些方法",阻止不完整的类被实例化,并在缺失实现时尽早报错(而非运行到调用才崩溃)。

14.1 基本用法

from abc import ABC, abstractmethod

class Animal(ABC):
    @abstractmethod
    def speak(self):
        pass                        # 子类必须实现

    @property
    @abstractmethod
    def legs(self):                # 抽象属性
        pass

class Dog(Animal):
    legs = 4
    def speak(self):
        return "woof"

# Animal()        # TypeError: Can't instantiate abstract class Animal with abstract method speak
dog = Dog()       # OK

14.2 虚拟子类(不通过继承注册)

register() 把一个不继承 ABC 的类登记为"虚拟子类",使其通过 isinstance 检查——实现"结构化兼容"而不强耦合。

from collections.abc import Sized

class MyCollection:
    def __len__(self):
        return 42

Sized.register(MyCollection)             # 注册为虚拟子类
print(isinstance(MyCollection(), Sized))  # True(虽未继承 Sized)

__subclasshook__:在 ABC 里重写它可实现自动鸭子类型识别(如 Iterable 只要对象有 __iter__ 就认为是它的子类)。collections.abc 大量使用此技巧。

14.3 常用 collections.abc 抽象基类

IterableIteratorSizedContainerSequenceMutableSequenceMappingMutableMappingSetMutableSetCallableHashable。用 isinstance(x, Sequence) 做接口检查比 hasattr(x,'__getitem__') 更可靠。

ABC vs Protocol(typing):ABC 是运行时强制(实例化时报错、可被 isinstance 检测);Protocol静态类型层面的结构化类型(mypy 检查,运行时无约束,零开销)。需要运行时强制用 ABC;只做静态检查用 Protocol。


15. collections 常用容器

标准 dict/list/tuple/set 之外,collections 提供一组高频专用容器。

容器 用途 关键方法/特性
namedtuple 轻量不可变记录 _asdict, _replace, _make, _fields
deque 双端队列,两端 O(1) appendleft, popleft, maxlen, 线程安全(append/pop)
Counter 计数/频次 most_common, elements, subtract, +/- 运算
defaultdict 带默认值工厂的 dict 访问缺失键自动调用工厂
ChainMap 多 dict 逻辑合并 不拷贝、查找按序、写只命中第一个
OrderedDict 保序 dict(3.7+ 内置 dict 已保序) move_to_end, popitem(last=)
from collections import namedtuple, deque, Counter, defaultdict, ChainMap, OrderedDict

# namedtuple:不可变数据记录(比 class 省内存,比 tuple 可读)
Point = namedtuple("Point", ["x", "y"])
p = Point(10, 20)
p.x, p[0]                 # 10, 10
p._asdict()               # {'x': 10, 'y': 20}
p._replace(x=99)          # Point(x=99, y=20) 返回新对象(不可变)
Point._fields             # ('x', 'y')
Point._make([1, 2])       # Point(x=1, y=2)

# deque:高效队列/栈(list 在头部插入是 O(n),deque 两端都 O(1))
dq = deque(maxlen=3)      # 定长:满了自动丢最旧
dq.extend([1, 2, 3, 4])   # deque([2, 3, 4])
dq.appendleft(0)
dq.popleft()

# Counter:统计神器
cnt = Counter("abracadabra")
cnt.most_common(2)        # [('a', 5), ('b', 2)]
cnt.elements()            # 展开成迭代器(按计数重复)
cnt.subtract("abra")      # 减法
Counter("aabb") + Counter("bbcc")   # Counter({'b':4,'a':2,'c':2})

# defaultdict:省去 "if key not in dd"
dd = defaultdict(list)
dd["users"].append("Alice")   # 不抛 KeyError
dd["score"]                   # [](默认工厂返回空列表)

# ChainMap:配置层层覆盖(不合并拷贝,查找按顺序)
defaults = {"host": "localhost", "port": 8080}
overrides = {"host": "prod"}
cm = ChainMap(overrides, defaults)
cm["host"]                    # 'prod'(先查 overrides)
cm["port"]                    # 8080(回退 defaults)
cm["host"] = "test"           # 只写入 overrides(第一个映射)

# OrderedDict:需依赖顺序语义时使用(如 LRU 实现)
od = OrderedDict()
od["a"] = 1; od["b"] = 2
od.move_to_end("a")           # 移到末尾

UserDict/UserList/UserString:需要继承并定制 dict/list/str 行为时,继承这些包装类比直接继承内置类型更安全(避免内置类型绕过重写的特殊方法)。


16. dataclasses

核心概念@dataclass 是一个类装饰器,自动为你生成 __init____repr____eq__ 等样板方法,让你用声明式语法写数据类,少敲大量重复代码。

16.1 基础

from dataclasses import dataclass, field

@dataclass
class User:
    name: str
    age: int = 0
    tags: list = field(default_factory=list)   # 可变默认值必须用 default_factory
    active: bool = True

    def greet(self):
        return f"Hi, {self.name}"

u = User("Alice", 30)
u2 = User("Bob")
print(u)                  # User(name='Alice', age=30, tags=[], active=True)
print(u == u2)            # False(自动 __eq__)
u.tags.append("vip")      # tags 是独立列表

致命陷阱:默认参数若直接写 tags: list = [],所有实例会共享同一个列表(和函数默认参数陷阱一致)。必须用 field(default_factory=list)

16.2 进阶选项

from dataclasses import dataclass, field, asdict, astuple

@dataclass(frozen=True)          # 不可变:禁止赋值,自动生成 __hash__
class FrozenPoint:
    x: float
    y: float

@dataclass(order=True)           # 自动生成 __lt__/__le__/__gt__/__ge__(按字段顺序比较)
class Rank:
    score: int = field(compare=True)
    name: str = field(compare=False)   # 比较时忽略 name

@dataclass
class Config:
    items: list = field(default_factory=list, repr=False)   # repr 中隐藏
    _cache: dict = field(default_factory=dict, init=False)  # 不在 __init__ 中

    def __post_init__(self):
        # __init__ 之后自动调用:做额外校验/推导
        if not self.items:
            raise ValueError("items 不能为空")

c = Config([1, 2])
asdict(c)                        # {'items': [1, 2], '_cache': {}}  转成普通 dict
astuple(c)                       # ([1, 2], {})

field 常用参数default(不可变默认值)、default_factory(工厂函数,处理可变默认值)、init(是否进 __init__)、repr(是否进 __repr__)、compare(是否参与比较)、kw_only(Python 3.10+,强制关键字参数)。

16.3 与 slots 结合 & 与 namedtuple 对比

@dataclass(slots=True)           # Python 3.10+:自动加 __slots__,省内存
class Efficient:
    x: int
    y: int
维度 namedtuple dataclass
可变性 不可变、可哈希 默认可变(frozen=True 才可哈希)
继承 不支持字段继承 支持字段继承
默认值 仅尾部可选 任意字段默认值、灵活
内存 最小(C 级元组) 普通对象;slots=True 后可接近
方法 _replace/_asdict 可自由定义方法

选型:仅做简单不可变记录 → namedtuple;需要可变性/默认值/继承/校验/序列化 → dataclass;需要运行时校验与序列化 → 考虑 pydantic


17. 类型注解 (typing)

核心概念:Python 是动态类型语言,但类型注解(type hints)让你标注变量/参数/返回值的预期类型。注解运行时被忽略(不强制、不改行为),主要服务于静态检查器(mypy/pyright)、IDE 提示、和文档。

17.1 基础注解与容器

from typing import Optional, Union, List, Dict, Tuple, Callable, Any

def greet(name: str) -> str:
    return f"Hello, {name}"

def get_config(key: str) -> Optional[int]:   # int | None
    return config.get(key)

def process(data: Union[int, str]) -> str:   # int | str
    return str(data)

users: List[str] = ["Alice", "Bob"]          # 注意:仅注解,运行时不检查
scores: Dict[str, int] = {}
handler: Callable[[int, int], str] = lambda a, b: f"{a}+{b}"

Python 3.9+ 推荐:直接用内置泛型 list[str]dict[str, int]tuple[float, float]X | NoneX | Y,无需从 typing 导入容器类型。上面写法是为兼容旧版本。

17.2 泛型(TypeVar / Generic)

from typing import TypeVar, Generic

T = TypeVar("T")
U = TypeVar("U", int, str)        # 约束:只能是 int 或 str
B = TypeVar("B", bound="Sequence") # 上界:必须是 Sequence 的子类

class Stack(Generic[T]):
    def __init__(self):
        self._items: list[T] = []
    def push(self, item: T) -> None:
        self._items.append(item)
    def pop(self) -> T:
        return self._items.pop()

s = Stack[int]()
s.push(1)

17.3 结构化类型:Protocol / TypedDict / Literal

from typing import Protocol, TypedDict, Literal, NewType, cast, overload

# Protocol:鸭子类型的静态表达(运行时零开销,不需继承)
class Flyable(Protocol):
    def fly(self) -> None: ...

def takeoff(obj: Flyable) -> None:
    obj.fly()                    # mypy 只要对象有 fly() 就通过

# TypedDict:字典的结构化注解(字段名+类型)
class UserInfo(TypedDict):
    name: str
    age: int

u: UserInfo = {"name": "Alice", "age": 30}

# Literal:精确字面值集合
def set_mode(mode: Literal["read", "write", "append"]) -> None: ...

# NewType:创建"名义上不同"的类型(运行时是同一类型,仅静态区分)
UserId = NewType("UserId", int)
uid = UserId(42)

# @overload:为同一函数声明多种签名(实现里仍是单个函数)
@overload
def to_str(x: int) -> str: ...
@overload
def to_str(x: bytes) -> str: ...
def to_str(x):
    return x.decode() if isinstance(x, bytes) else str(x)

17.4 运行时与静态

import typing
def f(a: "list[int]", b: int = 0): ...

typing.get_type_hints(f)          # 运行时读取注解(解析前向/字符串引用)
# 输出如 {'a': list[int], 'b': int} —— 注解被保留在 __annotations__

要点:注解存在 obj.__annotations__,但不做强制。要真正检查,需 mypy your_file.py 或编辑器实时检查。PEP 563(3.10+ 默认)让注解延迟求值(from __future__ import annotations 在所有版本可用),避免循环引用与启动开销。cast(T, val) 只告诉检查器"当成 T",运行时不改变 val


18. 并发模型对比

核心概念:Python 有三种并发武器,选择取决于任务是 CPU 密集还是 IO 密集,根因是 GIL(下节详述)。

18.1 三种模型

import threading, time

# threading —— IO 密集型(sleep/网络/磁盘等待会释放 GIL,其他线程能跑)
def io_work():
    time.sleep(1)                          # 释放 GIL
threads = [threading.Thread(target=io_work) for _ in range(10)]
[t.start() for t in threads]; [t.join() for t in threads]   # 约 1s

# multiprocessing —— CPU 密集型(多进程绕过 GIL,真并行)
import multiprocessing as mp
def compute():
    return sum(i*i for i in range(10**7))
with mp.Pool() as pool:
    pool.map(compute, range(4))            # 4 进程并行,约 1/4 时间

# asyncio —— 高并发 IO(单线程,协程协作式,开销最低)
import asyncio, aiohttp
async def fetch(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            return await r.text()

18.2 选型表

场景 推荐 原因
CPU 密集(计算) multiprocessing 多进程真并行,绕过 GIL
IO 密集(少量) threading 等待 IO 时释放 GIL,共享内存简单
IO 密集(高并发 1000+) asyncio 协程切换开销极低,单线程撑海量连接
混合 进程池 + 异步 CPU 与 IO 分离,进程跑计算、异步跑 IO
需要共享状态 + 简单 threading + Lock 但要注意竞态

18.3 共享状态的陷阱

import threading
counter = 0
lock = threading.Lock()

def inc():
    global counter
    for _ in range(100000):
        with lock:                 # 不加锁会出现竞态,结果 < 100000*N
            counter += 1

ts = [threading.Thread(target=inc) for _ in range(4)]
[t.start() for t in ts]; [t.join() for t in ts]
# 有锁时 counter == 400000

asyncio vs threading 心智模型threading抢占式(OS 决定切换点,需锁保护共享数据);asyncio协作式(只在 await 处切换,单线程无竞态,但不写 await 就占着循环)。asyncio 的优势在"海量连接"而非"单连接更快"。


19. 内存管理与 GIL

19.1 引用计数 + 分代垃圾回收

Python(CPython)用引用计数为主、分代 GC 兜底

  • 引用计数:每个对象存一个计数器,引用 +1 / -1。归零时立即回收。这是 Python 内存管理的主力,确定性高、开销小。
  • 循环引用:引用计数无法回收(A 引用 B、B 引用 A,计数永不为 0),由分代垃圾回收器(generation 0/1/2 三代)定期扫描兜底。
import sys, gc
a = [1, 2, 3]
print(sys.getrefcount(a))       # 查看当前引用计数(调用本身会 +1)
print(gc.get_threshold())       # (700, 10, 10) 各代阈值
print(gc.get_stats())           # 各代回收统计
gc.collect()                    # 手动触发全量回收

19.2 weakref — 打破循环引用

弱引用不增加引用计数,常用于缓存、观察者,避免对象被"本应回收"的引用钉住。

import weakref
class Node:
    def __init__(self, val):
        self.val = val
        self.parent = None

n = Node(1)
r = weakref.ref(n)              # 弱引用
print(r() is n)                 # True(对象还活着)
del n
print(r() is None)              # True(对象被回收,弱引用失效)

19.3 GIL(全局解释器锁)

定义:CPython 同一进程同一时刻只有一个线程能执行 Python 字节码。哪怕多核,多线程 Python 代码在 CPU 密集任务上无法真正并行

  • 为什么会这样:CPython 的内存管理(引用计数)不是线程安全的,GIL 是保护它的简单方案。
  • 何时有影响CPU 密集任务被 GIL 限制(多线程退化为串行);IO 密集任务在等待 IO 时会释放 GIL,多线程/异步仍能并发受益。
  • 绕过 GILmultiprocessing(多进程,真并行)、C 扩展(NumPy 等释放 GIL 做计算)、nogil 构建(PEP 703,Python 3.13+ 实验性"free-threaded"构建可禁用 GIL)。

19.4 id() / is 与对象身份

a = [1, 2]; b = [1, 2]
print(a is b)        # False(不同对象)
print(a == b)        # True(值相等)
print(id(a))         # 对象内存地址(进程内唯一标识)

is 比较身份(是否同一对象),== 比较值。常见误用:x is None(正确,单例)、x == None(不推荐)。注意小整数(-5~256)和短字符串会被 ** intern(缓存)**,所以 a = 256; b = 256; a is bTrue——这是实现细节,不要依赖。

19.5 不可变对象与内存视图

  • int/str/tuple/frozenset 不可变:看起来"修改"其实是创建新对象。
  • memoryview / bytearray:零拷贝地切片/操作缓冲,处理大二进制数据时省内存。

20. 性能分析与优化

核心原则先测量,再优化("过早优化是万恶之源")。用工具定位真正的瓶颈,而非凭直觉。

20.1 测量工具

import timeit, cProfile, pstats

# timeit:微基准(多次运行取均值,自动屏蔽噪声)
timeit.timeit("sum(range(1000))", number=10000)

# cProfile:函数级剖析(谁最耗时、调用几次)
cProfile.run("sum(range(10**7))", sort="tottime")
# CLI: python -m cProfile -s tottime your_script.py

# 更细:line_profiler(逐行)、memory_profiler(逐行内存)、py-spy(采样,生产无侵入)

cProfile 字段解读:tottime=函数自身耗时(不含子调用)、cumtime=含子调用总耗时、ncalls=调用次数、percall=单次平均。优先优化 tottime 高且 ncalls 大的函数。

20.2 常用优化手段(含原理)

import math

# 1) 局部变量缓存:循环内反复访问的全局/属性函数,先绑定到局部变量
#    避免每次都走 LEGB 查找链
_abs = abs
for x in range(10**6):
    _abs(x)

# 2) 列表推导 vs 循环 append:列表推导在 C 层循环,无每次 append 的属性查找
data = [x*2 for x in range(1000)]          # 快

# 3) " ".join() vs += 拼接:join 预计算总长一次分配;+= 字符串不可变,反复拷贝
parts = ["a", "b", "c"]
result = "".join(parts)                     # 快一个数量级

# 4) set/dict 查找 O(1) :成员检查用 set 而非 list
haystack = set(big_list)
if item in haystack: ...                    # 比 "item in big_list" 快 N 倍

# 5) @lru_cache:缓存重复/递归计算
from functools import lru_cache
@lru_cache(maxsize=256)
def expensive(n): ...

# 6) __slots__:大量实例时省内存(见第 4 节)

# 7) 用内置与 C 扩展:sum/min/max/sorted、NumPy 向量化远快于手写 Python 循环
total = sum(values)                          # 而非手写循环累加
手段 原理 适用
列表推导 C 层循环,无 append 属性查找 数据转换
str.join 预计算总长,单次分配 字符串拼接
set/dict 查找 哈希表 O(1) 成员检查
局部变量缓存 避免 LEGB 查找链 循环内函数调用
__slots__ __dict__,属性访问略快 大量实例
lru_cache 记忆化,避免重复计算 递归/频繁调用
内置/NumPy C 实现,向量化 数值计算

警告:大多数"优化"代价是可读性。+ 拼接一行字符串、普通 for 累加小幅数据,在绝大多数业务代码里完全够用。只针对 cProfile 指出的热点动手。必要时把热路径用 numba/Cython/C 扩展重写,而非在 Python 层抠细节。


21. 学习路线图

按由内到外、由基础到框架的顺序,建议分 9 步:

  1. 魔术方法 + 描述符(第 2、3 节)——先把对象模型吃透:知道属性从哪来、类怎么被创建、运算符如何被调用。这是理解后面一切的底座。
  2. 闭包 + 装饰器 + functools(第 5、6 节)——最常用、最立竿见影。闭包是装饰器的基础,@wraps、带参装饰器、缓存/重试是日常必备。
  3. 迭代器 + itertools(第 7、8 节)——理解 for 的底层,掌握惰性计算与组合工具,写高效数据流。
  4. 上下文管理器 + 生成器(第 9、10 节)——资源管理的标准范式(with),以及协程前身(yield/send)。
  5. asyncio(第 11 节)——从手写协程过渡到原生异步;吃透 Task/Queue/gather/run_in_executor,避免阻塞陷阱。
  6. super() + 元类 + ABC(第 12、13、14 节)——框架/ORM 的底层。以"理解其机制"为主,业务代码慎用元类,super() 单继承用法要熟练。
  7. collections + dataclasses(第 15、16 节)——选对数据结构、用声明式减少样板代码,少写字多做事。
  8. 类型注解 typing(第 17 节)——提升可读性与可维护性,配合 mypy/Pylance 做静态检查,大型项目收益显著。
  9. 并发 + 内存/GIL + 性能(第 18、19、20 节)——定位性能瓶颈、正确选型(线程/进程/异步),用 cProfile/timeit 量化收益。

一句话心智模型:Python 里"type 造出的对象(元类),方法/字段是描述符,协程是能挂起/恢复的调度单元(asyncio),__slots__ 是内存优化开关"。把这四点串起来,全部高级特性就不再是散点,而是一棵自洽的树。


Python 高级知识点完全指南(Markdown 版)| 2026-07 | 21 节 · 40+ 代码示例 · 9 幅图示 · 覆盖对象模型 / 函数式 / 元编程 / 执行模型 / 并发 / 数据结构 / 性能 | 建议在 Python 3.10+ 环境运行全部示例


如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
暫無回覆。
需要 登录 後方可回應,如果你還沒有帳號按這裡 注册