Python
已收录:本页完成领域概览规划清单中 Python 的 4 项内容(数据模型、装饰器/生成器/迭代器、类型注解与 Pydantic、NumPy/Pandas 速查)。示例标注运行环境,可直接复制验证。
本页按「语法要点 → 运行机制 → 示例 → 踩坑」组织:
数据模型:魔术方法与协议
对象模型总览(语法要点)
Python 通过「魔术方法」(dunder,双下划线)把语法糖映射到方法调用:x + y 实际调用 x.__add__(y),len(x) 调用 x.__len__。实现特定方法即「参与某个协议」。
# 运行环境: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 ≥ 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 ≥ 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 ≥ 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而不是返回None(for依赖它终止);- 属性名加前缀
_防与描述符名冲突(__set_name__常规用法);避免在__getattr__里访问不存在属性造成无限递归; functools.total_ordering只省写法不省性能,大量排序还是手写__lt__;- 不要在
__init__里做重计算(__init__每次实例化都跑),用functools.cached_property惰性计算。
装饰器、生成器与迭代器
迭代器与可迭代对象(运行机制)
- 可迭代(iterable):实现了
__iter__(返回迭代器)或__getitem__(旧协议),for/sum/list()可消费; - 迭代器(iterator):同时实现
__iter__(返回自身)与__next__(每次吐一个元素,耗尽抛StopIteration);迭代器是一次性的。
# 运行环境: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 ≥ 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 ≥ 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 ≥ 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 ≥ 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 ≥ 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 ≥ 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 ≥ 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 ≥ 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)(省内存)。
静态检查工程配置(工程实践)
# 运行环境: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 ≥ 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_name、SecretStr(脱敏输出)、field_validator(mode='before')(在类型转换前处理原始值)。
# 运行环境: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_validator;Config类改为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:数组创建与形状
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 增加新轴索引、布尔筛选与广播:
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
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'] > 21groupby 聚合、透视与合并:
# 分组聚合:分组键 → 聚合函数(可多列多函数)
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) # 纵向堆叠缺失值与时间序列:
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' 已弃用)# 文件读写(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()实例,避免全局状态污染可复现实验。
状态与参考
- 状态:已收录(2026-09-02 完成领域概览规划的 4 项主题)。
- 运行环境:Python ≥ 3.10;Pydantic 示例需
pip install pydantic;数据科学示例需pip install numpy pandas。 - 参考:Python 官方文档、Pydantic 文档、Pandas 用户指南、NumPy 速查表。
下一步
- [ ] 把数据科学相关实操(真实 CSV 清洗/聚合)沉淀为可复用示例
- [ ] 跟进 Pydantic 与 FastAPI 的组合实践(请求模型、响应模型、OpenAPI 生成)
- [ ] 补充 pandas 高阶主题:窗口函数
rolling/expanding、groupby.transform、性能调优(eval/query、分块读取)
写作规范与页面规划请参阅领域概览。