Skip to content

Python

已收录:本页完成领域概览规划清单中 Python 的 4 项内容(数据模型、装饰器/生成器/迭代器、类型注解与 Pydantic、NumPy/Pandas 速查)。示例标注运行环境,可直接复制验证。

本页按「语法要点 → 运行机制 → 示例 → 踩坑」组织:

  1. 数据模型:魔术方法与协议
  2. 装饰器、生成器与迭代器
  3. 类型注解与 Pydantic
  4. NumPy / Pandas 常用操作速查

数据模型:魔术方法与协议

对象模型总览(语法要点)

Python 通过「魔术方法」(dunder,双下划线)把语法糖映射到方法调用:x + y 实际调用 x.__add__(y)len(x) 调用 x.__len__。实现特定方法即「参与某个协议」。

python
# 运行环境:Python ≥ 3.10
class Vec:
    def __init__(self, x, y):
        self.x, self.y = x, y

    def __repr__(self):              # repr() 与调试输出,尽量能还原对象
        return f'Vec({self.x}, {self.y})'

    def __str__(self):               # str()/print 优先,缺省回落 __repr__
        return f'({self.x}, {self.y})'

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

    def __eq__(self, other):         # ==;注意定义后默认不可哈希(见踩坑)
        return isinstance(other, Vec) and (self.x, self.y) == (other.x, other.y)

v1, v2 = Vec(1, 2), Vec(3, 4)
print(v1 + v2)   # (4, 6)
print(repr(v1))  # Vec(1, 2)

核心协议速查表

类别方法触发语法说明
对象生命周期__new__ / __init__C(...)前者建对象(不可变类型/单例用),后者初始化
表示__repr__ / __str__repr(x) / str(x)__repr__ 面向开发者
相等与哈希__eq__ / __hash__ / __lt__== / hash() / 排序比较全套实现一个即可用 functools.total_ordering 补全
数值__add__ / __sub__ / __mul__ / __truediv__ / __iadd__ / __neg__+ - * / += -x就地版 __iadd__ 返回自身则原地
容器__len__ / __getitem__ / __setitem__ / __delitem__ / __contains__len(x) / x[k] / k in x实现 __getitem__ 即成为可订阅对象
迭代__iter__ / __next__for x in obj见第二节迭代器协议
上下文管理__enter__ / __exit__with obj as y:资源管理,__exit__ 返回真则吞异常
可调用__call__obj(args)让实例像函数
属性__getattr__ / __setattr__ / __getattribute__ / __set_name__属性访问/赋值前者只在查找失败时触发
布尔与长度__bool__bool(x) / if x缺省回落到 __len__

自定义容器与上下文管理器的完整示例:

python
# 运行环境:Python ≥ 3.10
class Timer:
    """上下文管理器:with 进入/退出时打印耗时"""
    def __init__(self, name):
        self.name = name

    def __enter__(self):
        import time
        self._t0 = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc, tb):
        import time
        print(f'{self.name} 耗时 {time.perf_counter() - self._t0:.4f}s')
        return False          # False = 不吞异常;True = 吞掉异常

with Timer('求和'):
    sum(range(10_000_000))
python
# 运行环境:Python ≥ 3.10
class Range:
    """最小可迭代容器:实现 __getitem__ 即可被 for 遍历(旧式迭代协议)"""
    def __init__(self, n): self.n = n
    def __getitem__(self, i):
        if i >= self.n: raise IndexError
        return i * 10
for x in Range(3):
    print(x, end=' ')   # 0 10 20

描述符(运行机制)

property、类方法、ORM 字段的底层是描述符协议__get__/__set__):实例属性访问会先查类上的描述符。

python
# 运行环境:Python ≥ 3.10
class Positive:
    """数据描述符:校验每次赋值非负"""
    def __set_name__(self, owner, name):   # 3.6+:类创建时自动注入属性名
        self.name = '_' + name
    def __get__(self, obj, owner=None):
        return getattr(obj, self.name)
    def __set__(self, obj, value):
        if value < 0:
            raise ValueError(f'{self.name[1:]} 必须非负,收到 {value}')
        setattr(obj, self.name, value)

class Account:
    balance = Positive()
    def __init__(self, balance): self.balance = balance

a = Account(100)
a.balance += 50
print(a.balance)        # 150
# a.balance = -1        # ValueError:balance 必须非负

踩坑记录

  • 定义 __eq__ 后默认不可哈希(__hash__ 被置 None):可变对象本就该不可哈希;不可变对象记得显式补 __hash__
  • 可变对象不要实现 __hash__(否则放进 set/dict 后内容变了 hash 失效);
  • __getitem__ 越界要抛 IndexError 而不是返回 Nonefor 依赖它终止);
  • 属性名加前缀 _ 防与描述符名冲突(__set_name__ 常规用法);避免在 __getattr__ 里访问不存在属性造成无限递归;
  • functools.total_ordering 只省写法不省性能,大量排序还是手写 __lt__
  • 不要在 __init__ 里做重计算(__init__ 每次实例化都跑),用 functools.cached_property 惰性计算。

装饰器、生成器与迭代器

迭代器与可迭代对象(运行机制)

  • 可迭代(iterable):实现了 __iter__(返回迭代器)或 __getitem__(旧协议),for/sum/list() 可消费;
  • 迭代器(iterator):同时实现 __iter__(返回自身)与 __next__(每次吐一个元素,耗尽抛 StopIteration);迭代器是一次性的。
python
# 运行环境:Python ≥ 3.10
class CountDown:
    """迭代器示例:倒计时"""
    def __init__(self, n): self.n = n
    def __iter__(self): return self
    def __next__(self):
        if self.n <= 0: raise StopIteration
        self.n -= 1
        return self.n + 1

print(list(CountDown(3)))   # [3, 2, 1]

生成器(语法要点)

yield 的函数调用后返回生成器对象(本身就是迭代器):代码惰性执行,每次 __next__ 跑到下一个 yield 暂停,函数局部变量保留在暂停点。

python
# 运行环境:Python ≥ 3.10
def fib():
    a, b = 0, 1
    while True:
        yield a          # 暂停并吐出 a
        a, b = b, a + b

g = fib()
print([next(g) for _ in range(7)])   # [0, 1, 1, 2, 3, 5, 8]

# 惰性求值:处理超大文件/无限流时不把数据全载入内存
def read_lines(path):
    with open(path, encoding='utf-8') as f:
        for line in f:
            yield line.strip()
# for line in read_lines('big.log'): ...  —— 一次只占一行内存

生成器通信(send/throw/close)与委托(yield from):

python
# 运行环境:Python ≥ 3.10
def echo():
    while True:
        received = yield 'ready'   # send 进来的值成为 yield 表达式结果
        yield f'got {received}'

e = echo()
print(next(e))            # 'ready'
print(e.send('hi'))       # 'got hi'

# yield from 委托给子生成器,透传 send/throw
def outer():
    yield 'start'
    yield from fib()      # 转交控制权

装饰器(运行机制)

装饰器是「接收函数、返回函数」的高阶函数,@deco 等价于 f = deco(f)。由于包装后 __name__/__doc__ 会丢失,务必用 functools.wraps 透传元信息:

python
# 运行环境:Python ≥ 3.10
import functools, time

def timed(fn):
    @functools.wraps(fn)          # 拷贝 __name__、__doc__、__wrapped__ 等
    def wrapper(*args, **kwargs):
        t0 = time.perf_counter()
        try:
            return fn(*args, **kwargs)
        finally:
            print(f'{fn.__name__} 耗时 {time.perf_counter() - t0:.4f}s')
    return wrapper

@timed
def slow_work(n):
    """计算 0..n 的和"""
    return sum(range(n))

slow_work(10_000_000)
print(slow_work.__name__, slow_work.__doc__)  # slow_work / 计算 0..n 的和(未被吞掉)

带参数的装饰器 = 再包一层工厂;也用类实现(__call__):

python
# 运行环境:Python ≥ 3.10
def retry(times=3, delay=0.1):
    def deco(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            import time
            last = None
            for _ in range(times):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    last = e
                    time.sleep(delay)
            raise last
        return wrapper
    return deco

@retry(times=2)
def flaky():
    raise ConnectionError('网络抖动')
# flaky()  # 重试 2 次后抛 ConnectionError

内置装饰器一览(重点理解 @property@classmethod 的区别):

装饰器作用典型场景
@property方法变只读属性,配套 @x.setter/@x.deleter计算属性、防改的对外字段
@classmethod首个参数为类 cls,可用 C.method()/子类继承备选构造函数(from_dict
@staticmethod与类无关的普通函数工具函数(不强依赖类)
@functools.cached_property惰性计算并缓存(实例级)昂贵且只算一次的属性
@functools.lru_cache(maxsize=None)按参数缓存函数结果纯函数加速(参数须可哈希)
@dataclass按注解自动生成 __init__/__repr__/__eq__数据载体(见第三节)
@contextlib.contextmanager用生成器实现上下文管理器免手写 __enter__/__exit__
python
# 运行环境:Python ≥ 3.10
import contextlib

@contextlib.contextmanager
def chdir(path):
    """临时切换工作目录,退出时自动还原(yield 前后即 __enter__/__exit__ 逻辑)"""
    import os
    old = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old)
# with chdir('/tmp'): run_build()   —— 构建结束 cwd 已还原

itertools 高频工具:

python
# 运行环境:Python ≥ 3.10
import itertools as it
print(it.pairwise('abcde'))                      # ('a','b') ('b','c') ...(3.10+)
print(list(it.islice(it.count(10), 3)))          # [10, 11, 12] 无限计数取前 3
print(list(it.product('ab', repeat=2)))          # 笛卡尔积
print(list(it.combinations('abc', 2)))           # 组合
print(list(it.chain([1, 2], ['x'])))             # [1, 2, 'x']
# groupby 需先排序;zip_longest 补位

踩坑记录

  • 生成器不能复用g = (x for x in ...) 用完后 list(g) 第二次为空,需重建;
  • 装饰器忘了 @functools.wraps:堆栈信息、文档、序列化(__name__)全乱;
  • 带参装饰器少一层括号是最常见的运行时报错(@retry@retry(2) 二选一,统一用带括号形式最省心);
  • lru_cache 的默认参数若为可变对象(list/dict)会直接报 TypeError:全部不可哈希;
  • 在生成器函数里写 return 是「结束」,返回值不会作为最后一个元素吐出;
  • 列表推导与生成器表达式只差括号:(x for x in ...) 是惰性、[x for x in ...] 立即物化,超大输入别用前者结果复用多次。

类型注解与 Pydantic

类型注解基础(语法要点)

注解只提供「静态检查」信息(运行时默认不校验);配合 mypy / pyright 或 pydantic 才有实际约束。语法随版本演进:3.9 起内置泛型(list[int] 而非 typing.List),3.10 起 X | None 替代 Optional[X]

python
# 运行环境:Python ≥ 3.10(检查需 mypy / pyright)
from typing import Literal, Protocol, TypedDict, TypeAlias, runtime_checkable

def greet(name: str, times: int = 1) -> str:   # 参数与返回注解
    return f'hi {name}' * times

userId: int = 42

# 泛型容器
scores: dict[str, list[int]] = {'alice': [1, 2]}

# 字面量联合:限定取值
Level: TypeAlias = Literal['debug', 'info', 'error']
def log(lv: Level) -> None: ...

# 结构化类型:鸭子类型的静态化(Protocol)
@runtime_checkable
class Sized(Protocol):
    def __len__(self) -> int: ...

def total(s: Sized) -> int: return sum(range(len(s)))
total([1, 2])     # ✅ list 有 __len__
# total(123)      # ❌ int 无 __len__(mypy 报错)

# TypedDict:dict 的键类型约束(数据源为 JSON 时常用)
class Payload(TypedDict, total=False):
    id: int
    name: str
p: Payload = {'id': 1, 'name': 'x'}   # ✅

泛型函数/类与运行时取注解:

python
# 运行环境:Python ≥ 3.10
from typing import TypeVar, Generic

T = TypeVar('T')

def first(items: list[T]) -> T:      # 泛型函数:返回类型与入参元素一致
    return items[0]

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

# 运行时拿到注解(默认字符串需 evaluate,或开启 __future__ annotations 再转)
from typing import get_type_hints
def f(x: int) -> str: ...
# get_type_hints(f)  # {'x': int, 'return': str}

进阶写法速查:Self(3.11,链式方法返回自身类型)、type X = ...(3.12 类型别名)、NewType(区分同名不同义的类型)、NamedTuple(轻量数据类)、dataclass(slots=True)(省内存)。

静态检查工程配置(工程实践)

toml
# 运行环境:Python ≥ 3.10 —— pyproject.toml(节选,配合 mypy/pyright 使用)
[tool.mypy]
python_version = "3.12"
strict = true
exclude = ["tests/", "migrations/"]

[tool.pyright]
pythonVersion = "3.12"
typeCheckingMode = "strict"

要点:库/脚本入口统一开 strict;遗留代码用 # type: ignore[code] 局部豁免而非关全局;# type: ignore 后跟具体错误码便于审计。

Pydantic(v2)

Pydantic 在运行时解析并校验数据(典型场景:API 请求体、配置、外部 JSON),v2 的校验内核为 Rust 实现(pydantic-core),比 v1 快数倍到数十倍。

python
# 运行环境:Python ≥ 3.10,pip install pydantic>=2.5
from typing import Literal
from pydantic import BaseModel, Field, ValidationError, field_validator, ConfigDict

class Item(BaseModel):
    model_config = ConfigDict(extra='forbid')   # 拒绝未知字段

    name: str
    price: float = Field(gt=0)                  # 声明式约束
    tag: Literal['a', 'b'] = 'a'

    @field_validator('name')
    @classmethod
    def no_blank(cls, v: str) -> str:           # 自定义校验(函数式,复用字段校验)
        if not v.strip(): raise ValueError('name 不能为空')
        return v.strip()

class Order(BaseModel):
    items: list[Item]                            # 嵌套模型自动递归校验
    total: float | None = None

    @model_validator(mode='after')
    def _fill_total(self):
        self.total = sum(i.price for i in self.items)
        return self

# 校验失败给出可读错误
try:
    Order(items=[{'name': '', 'price': -1, 'tag': 'x'}])
except ValidationError as e:
    print(e)
# 3 validation errors for Order
#  items -> 0 -> name  / 不能为空
#  items -> 0 -> price / 大于 0
#  items -> 0 -> tag   / tag 不在预期值中 ...

实际用法要点:Order.model_validate(payload)(对象/JSON 输入均可用)、model_dump() / model_dump_json()(输出)、model_validate_json(s)(字符串)、错误用 e.errors() 拿到结构化错误。配置项:strict=True(严格模式不做隐式转换)、alias/validation_alias(映射外部字段名)、populate_by_nameSecretStr(脱敏输出)、field_validator(mode='before')(在类型转换前处理原始值)。

python
# 运行环境:Python ≥ 3.10,pydantic ≥ 2.5
# 与 FastAPI 搭配时:函数参数直接标模型即可自动校验/生成 OpenAPI
from pydantic import BaseModel, SecretStr

class Cfg(BaseModel):
    db_url: SecretStr
    retries: int = 3

c = Cfg(db_url='postgres://u:p@h/db')
print(c.model_dump())   # {'db_url': SecretStr('**********'), 'retries': 3}(密文已脱敏)

踩坑记录

  • 注解默认不校验:接口数据必须过 pydantic/自写校验,别指望 mypy 保运行时安全;
  • Pydantic v1 → v2 破坏性变更:.dict()/.json() 改为 model_dump()/model_dump_json()@validator 改为 @field_validatorConfig 类改为 model_config = ConfigDict(...)parse_obj 改为 model_validate
  • 可变默认值(def f(x=[]))仍是被 mypy 标红的经典错误:用 x: list | None = None + 内部 x = x or []
  • Optional[X] 不代表「允许缺失字段」:Pydantic 里 x: int | None = None 才是「可缺省可空」;
  • Field(gt=0) 校验失败报错在赋值时(模型创建即校验),配置 validate_assignment=True 才能拦住后续属性赋值;
  • pandas/numpy 对象做类型注解别写 list[DataFrame] 意图错误:用 pd.DataFrame/numpy.ndarray 原生类型,或用 Sequence[pd.DataFrame] 表达容器语义。

NumPy / Pandas 常用操作速查

运行环境:Python ≥ 3.10,pip install numpy pandas(建议 numpy ≥ 1.26 / pandas ≥ 2.1)。以下均为可执行片段。

NumPy:数组创建与形状

python
import numpy as np
a = np.arange(12).reshape(3, 4)      # [[ 0  1  2  3] ... [ 8  9 10 11]]
z = np.zeros((2, 2), dtype=float)    # 全 0
r = np.random.default_rng(42)        # 新版推荐:全局 np.random 已被默认 RNG 取代
x = r.normal(size=(3, 3))            # 标准正态

a.shape, a.ndim, a.dtype             # (3, 4)  2  int64
a.T                                   # 转置
a.reshape(-1, 2)                      # 自动推导行数,变 6×2
a.ravel()                             # 展平成一维视图(copy=False 时共享内存)
b = np.concatenate([a, a], axis=0)    # 拼接 6×4;np.stack 增加新轴

索引、布尔筛选与广播:

python
a = np.arange(12).reshape(3, 4)
print(a[1, 2])          # 6 标量
print(a[:, 1])          # 整列:[1 5 9]
print(a[a % 2 == 0])    # 布尔掩码筛选:[ 0  2  4 ... 10]
print(a[a > 8])         # [ 9 10 11]

# 广播:小数组自动扩展到大数组(维度从右对齐,缺失补 1 或相等)
v = np.array([10, 20, 30, 40])
print(a + v)            # 每行加 v(3×4 与 4 对齐成功)
# print(a + np.array([1, 2]))  # 形状不匹配会广播失败(ValueError)

# 向量化聚合(避免 Python 循环)
a.mean(), a.sum(axis=0), a.max(axis=1)   # 标量 / 每列和 / 每行最大值
np.percentile(a, [25, 75])               # 分位数
np.clip(a, 2, 9)                         # 截断

Pandas:Series 与 DataFrame

python
import pandas as pd

df = pd.DataFrame({
    'dept': ['dev', 'ops', 'dev', 'ops', 'dev'],
    'name': ['a', 'b', 'c', 'd', 'e'],
    'salary': [20, 22, 25, 30, 18],
    'joined': pd.to_datetime(['2023-01-01', '2023-02-15', '2024-01-10', '2024-06-01', '2025-03-01']),
})
df.info()          # 列、非空计数、dtype 总览
df.describe()      # 数值列统计量
df['dept'].value_counts()   # 分组计数

# 选择:loc 标签 / iloc 位置;行筛选用布尔掩码
df.loc[df['salary'] > 20, ['name', 'salary']]
df.iloc[1:3]                              # 第 2、3 行

# 新增列:向量化(优先,勿用逐行循环)
df['salary_k'] = df['salary'] / 1000
df['senior'] = df['salary'] > 21

groupby 聚合、透视与合并:

python
# 分组聚合:分组键 → 聚合函数(可多列多函数)
df.groupby('dept')['salary'].agg(['mean', 'max', 'count'])

# 透视表:行=dept,列=年份,值=薪资均值
df['year'] = df['joined'].dt.year
pd.pivot_table(df, index='dept', columns='year', values='salary', aggfunc='mean')

# 长宽表转换:melt(宽→长)
wide = pd.DataFrame({'name': ['a', 'b'], 'q1': [1, 2], 'q2': [3, 4]})
long = wide.melt(id_vars='name', var_name='quarter', value_name='score')

# 合并:merge 类似 SQL join;concat 按轴堆叠
left = pd.DataFrame({'id': [1, 2], 'x': ['L1', 'L2']})
right = pd.DataFrame({'id': [1, 3], 'y': ['R1', 'R3']})
pd.merge(left, right, on='id', how='left')     # 左连接
pd.concat([left, right], ignore_index=True)    # 纵向堆叠

缺失值与时间序列:

python
s = pd.Series([1, None, 3, float('nan')])
s.isna().sum()          # 缺失计数(None 与 NaN 都算缺失)
df2 = df.dropna(subset=['salary'])          # 删指定列含缺失的行
df['salary'] = df['salary'].fillna(df['salary'].median())  # 中位数填充

# 时间序列重采样:先设为索引
ts = df.set_index('joined')['salary']
# 按月求和(示例数据粒度细到日,此处仅示意 API)
# monthly = ts.resample('ME').sum()   # 'ME' = 月末(2.x 新写法;旧版 'M' 已弃用)
python
# 文件读写(CSV/Excel/Parquet 是数据工程最常见三件套)
# df.to_csv('out.csv', index=False)          # index=False 不写行号列
# df = pd.read_csv('out.csv')
# df.to_parquet('out.parquet')               # 需 pyarrow;二进制列式,更快更省

数据科学踩坑

  • 视图 vs 副本df[df.x > 1]['y'] = 0 链式赋值会触发 SettingWithCopyWarning 且可能不生效——用 .loc[条件, 列] = 值 单步写;
  • NumPy 的 copy=False 视图共享内存:改动视图会改原数组,不确定时显式 .copy()
  • 布尔掩码里用 and/or 报错:数组逻辑请用 &/|每侧都要加括号df[(a) & (b)]);
  • 读 CSV 注意编码(encoding='utf-8')与类型(parse_dates=['joined']dtype),大文件用 chunksize 或换 parquet;
  • apply 慢于向量化:能写广播/groupby.transform 就别逐行 apply(万行以上差距明显);
  • pandas 2.x 弃用 inplace=True 系列(inplace 参数将在未来移除),推荐 df = df.dropna(...) 赋值式写法;
  • 随机数:新版请用 np.random.default_rng() 实例,避免全局状态污染可复现实验。

状态与参考

下一步

  • [ ] 把数据科学相关实操(真实 CSV 清洗/聚合)沉淀为可复用示例
  • [ ] 跟进 Pydantic 与 FastAPI 的组合实践(请求模型、响应模型、OpenAPI 生成)
  • [ ] 补充 pandas 高阶主题:窗口函数 rolling/expandinggroupby.transform、性能调优(eval/query、分块读取)

写作规范与页面规划请参阅领域概览

基于 VitePress 构建 · 内容以知识共享方式沉淀