服务端6月19日 12:15
Python 函数式编程怎么用才适合实际项目？## Python 函数式编程先解决什么问题
在 Python 里谈函数式编程，不是要把所有代码都写成一串看不懂的 `lambda`。它更适合处理这类问题：一批数据进来，经过过滤、转换、排序、聚合，最后得到一个新结果。
比如清洗接口返回的订单、统计日志里的错误类型、把配置项按规则合并。只要中间步骤能拆成几个独立的小函数，函数式写法就能让数据流向更清楚，也更容易测试。
Python 不是纯函数式语言，所以不用排斥循环、类和可变对象。更实际的做法是：在关键的数据处理逻辑里多用纯函数、少改共享状态，必要时用 `map`、`filter`、`reduce`、生成器和装饰器把重复逻辑收起来。
## 两个基础习惯：纯函数和不可变数据
纯函数有两个特点：同样的输入总是得到同样的输出；函数内部不修改外部状态。这个习惯看起来朴素，但在排查线上问题时很省心，因为你不用猜某个全局变量是不是被别处改过。
```python
# 纯函数：只依赖参数，不改外部状态
def add_tax(price, rate):
return round(price * (1 + rate), 2)
print(add_tax(100, 0.06)) # 106.0
print(add_tax(100, 0.06)) # 106.0
# 非纯函数：依赖并修改外部状态
total = 0
def add_to_total(amount):
global total
total += amount
return total
```
不可变数据的意思不是所有地方都必须用元组，而是尽量别在函数里偷偷改传进来的对象。尤其是列表、字典这种可变对象，修改前最好明确创建新对象。
```python
# 更稳妥：返回新列表
def append_item(items, item):
return items + [item]
original = [1, 2, 3]
new_items = append_item(original, 4)
print(original) # [1, 2, 3]
print(new_items) # [1, 2, 3, 4]
```
写业务代码时，这个区别很常见。比如函数接收一份配置，如果直接往里面塞默认值，调用方后面可能拿到一份已经被改过的配置；返回新配置就清楚得多。
## map、filter、reduce、sorted 分别适合什么场景
这几个函数是 Python 函数式编程里最常见的工具。别把它们当成必须使用的写法，先看语义是否合适。
| 工具 | 适合场景 | 常见替代 |
|---|---|---|
| `map` | 每个元素都做同一种转换 | 列表推导式 |
| `filter` | 按条件保留一部分元素 | 带 `if` 的列表推导式 |
| `reduce` | 把多个值折叠成一个值 | `sum`、`max`、普通循环 |
| `sorted` | 排序并返回新列表 | `list.sort()` 会原地修改 |
`map` 适合表达“逐个转换”：
```python
numbers = [1, 2, 3, 4, 5]
squared = list(map(lambda x: x ** 2, numbers))
print(squared) # [1, 4, 9, 16, 25]
left = [1, 2, 3]
right = [4, 5, 6]
print(list(map(lambda x, y: x + y, left, right))) # [5, 7, 9]
```
`filter` 适合表达“筛掉不需要的”：
```python
numbers = range(1, 11)
even_numbers = list(filter(lambda x: x % 2 == 0, numbers))
print(even_numbers) # [2, 4, 6, 8, 10]
words = ["apple", "banana", "cherry", "date"]
long_words = list(filter(lambda word: len(word) > 5, words))
print(long_words) # ['banana', 'cherry']
```
`reduce` 适合做累积，不过要谨慎。能用 `sum()`、`max()`、`min()` 表达清楚时，不必强行用 `reduce`。
```python
from functools import reduce
numbers = [1, 2, 3, 4, 5]
print(reduce(lambda x, y: x + y, numbers)) # 15
print(reduce(lambda x, y: x * y, numbers)) # 120
print(reduce(lambda x, y: x + y, numbers, 10)) # 25，初始值为 10
```
`sorted` 的好处是不会改原列表，配合 `key` 很适合处理对象或字典列表。
```python
students = [
{"name": "Alice", "age": 25},
{"name": "Bob", "age": 20},
{"name": "Charlie", "age": 30},
]
by_age = sorted(students, key=lambda item: item["age"])
print(by_age[0]) # {'name': 'Bob', 'age': 20}
```
## lambda 要短，复杂逻辑交给命名函数
`lambda` 是匿名函数，适合一眼能看懂的小逻辑，比如取字段、简单计算、简单条件。它只能写表达式，不能写多行语句。
```python
students = [("Alice", 25), ("Bob", 20), ("Charlie", 30)]
sorted_students = sorted(students, key=lambda item: item[1])
print(sorted_students)
```
如果逻辑开始变长，就别硬塞进 `lambda`。命名函数不仅更好读，也方便单独测试。
```python
# 不推荐：条件嵌套太多
score_to_level = lambda score: "A" if score >= 90 else "B" if score >= 80 else "C"
# 更清楚：写成普通函数
def score_to_level(score):
if score >= 90:
return "A"
if score >= 80:
return "B"
return "C"
```
一个简单判断：如果你需要回头数括号，或者要解释这个 `lambda` 到底在做什么，那它就该变成普通函数了。
## 列表推导式和生成器表达式更有 Python 味
很多时候，列表推导式比 `map` 和 `filter` 更直观。尤其是转换和过滤同时出现时，它读起来更接近自然语言。
```python
numbers = [1, 2, 3, 4, 5]
even_squared = [x ** 2 for x in numbers if x % 2 == 0]
print(even_squared) # [4, 16]
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
flattened = [item for row in matrix for item in row]
print(flattened) # [1, 2, 3, 4, 5, 6, 7, 8, 9]
```
数据量大时，生成器表达式更合适。它不会一次性把所有结果塞进内存，而是用到一个算一个。
```python
large_squares = (x ** 2 for x in range(1_000_000))
for value in large_squares:
if value > 100:
print(value)
break
```
处理日志、文件行、分页接口时，生成器很实用。它让“数据可能很多”这件事不必一开始就变成内存压力。
## 装饰器：把横切逻辑从业务函数里拿走
装饰器本质上也是高阶函数：接收一个函数，返回一个新函数。它适合处理日志、鉴权、重试、计时、缓存这类“很多函数都需要，但又不属于核心业务”的逻辑。
```python
from functools import wraps
import time
def timer(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
cost = time.perf_counter() - start
print(f"{func.__name__} took {cost:.4f}s")
return result
return wrapper
@timer
def load_items():
return [x for x in range(10000)]
```
这里的 `@wraps` 很重要，它会保留原函数的 `__name__`、文档字符串等元数据。没有它，调试、日志和一些框架反射逻辑可能会看到一堆 `wrapper`。
带参数的装饰器也很常见，比如做简单重试：
```python
from functools import wraps
def retry(times):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_error = None
for _ in range(times):
try:
return func(*args, **kwargs)
except Exception as error:
last_error = error
raise last_error
return wrapper
return decorator
```
## partial：提前固定一部分参数
`functools.partial` 可以把一个函数的部分参数先固定住，得到一个更具体的新函数。它比自己写一堆薄封装函数更省事。
```python
from functools import partial
def power(base, exponent):
return base ** exponent
square = partial(power, exponent=2)
cube = partial(power, exponent=3)
print(square(5)) # 25
print(cube(5)) # 125
```
在项目里，它常用于给通用函数预置上下文。
```python
from functools import partial
def format_message(level, service, message):
return f"[{level}] {service}: {message}"
api_error = partial(format_message, "ERROR", "payment-api")
print(api_error("timeout")) # [ERROR] payment-api: timeout
```
如果参数越来越多，`partial` 也可能让人迷糊。这个时候改成一个带清晰名字的函数，反而更直接。
## 用小函数拼出数据处理管道
函数式写法最舒服的场景，是把数据处理拆成几个小步骤。每个函数只做一件事，输入输出都清楚，组合起来就是一条管道。
```python
from functools import reduce
orders = [
{"id": 1, "status": "paid", "amount": 120, "country": "CN"},
{"id": 2, "status": "cancelled", "amount": 80, "country": "US"},
{"id": 3, "status": "paid", "amount": 260, "country": "CN"},
]
def is_paid(order):
return order["status"] == "paid"
def to_amount(order):
return order["amount"]
def add(left, right):
return left + right
paid_orders = filter(is_paid, orders)
paid_amounts = map(to_amount, paid_orders)
total_amount = reduce(add, paid_amounts, 0)
print(total_amount) # 380
```
上面这段可以继续写得更 Pythonic：
```python
total_amount = sum(
order["amount"]
for order in orders
if order["status"] == "paid"
)
```
这不是说 `map/filter/reduce` 不好，而是 Python 里有多种表达方式。短逻辑用推导式，复用逻辑拆成命名函数，复杂聚合用普通循环，都很正常。
## 函数组合和柯里化适合少量使用
函数组合就是把多个函数接起来，前一个函数的输出作为后一个函数的输入。它能让处理流程更集中。
```python
def compose(*functions):
def inner(value):
result = value
for func in reversed(functions):
result = func(result)
return result
return inner
def add_one(x):
return x + 1
def multiply_two(x):
return x * 2
def square(x):
return x ** 2
pipeline = compose(square, multiply_two, add_one)
print(pipeline(3)) # 64
```
柯里化是把多参数函数变成一连串单参数函数。它在一些函数式语言里很常见，在 Python 里偶尔有用，但不要为了形式感到处写。
```python
def curry(func):
def curried(*args):
if len(args) >= func.__code__.co_argcount:
return func(*args)
return lambda *more: curried(*(args + more))
return curried
@curry
def add(a, b, c):
return a + b + c
print(add(1)(2)(3)) # 6
```
实际项目里，`partial` 往往比通用柯里化更容易被团队接受。组合和柯里化适合用在边界清晰的数据转换上，不适合把普通业务流程绕成谜题。
## 记忆化：让重复计算少做几次
如果一个函数是纯函数，而且同样的参数会反复出现，就可以考虑记忆化。Python 标准库里的 `lru_cache` 已经够用。
```python
from functools import lru_cache
@lru_cache(maxsize=128)
def fibonacci(n):
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
print(fibonacci(100))
print(fibonacci.cache_info())
```
手写一个简化版也不难：
```python
def memoize(func):
cache = {}
def wrapper(*args):
if args not in cache:
cache[args] = func(*args)
return cache[args]
return wrapper
```
不过缓存不是免费午餐。参数必须可哈希，缓存会占内存，数据有时效性时还要考虑失效策略。像“按用户权限查询结果”这类函数，缓存前要先确认参数是否完整表达了影响结果的所有条件。
## 函数式写法带来的几个实际好处
第一是可测试。纯函数不依赖数据库、全局变量、当前时间，单测只需要准备输入和断言输出。
```python
def calculate_discount(price, discount_rate):
return price * (1 - discount_rate)
assert calculate_discount(100, 0.2) == 80
assert calculate_discount(50, 0) == 50
```
第二是可预测。函数不偷偷修改外部对象，调用前后状态更容易判断。多人协作时，这比少写两行代码更重要。
第三是更容易并行。没有共享状态的函数，放进线程池或进程池时少很多锁和竞态问题。
```python
from concurrent.futures import ThreadPoolExecutor
def process_item(item):
return item ** 2
items = list(range(1000))
with ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(process_item, items))
```
第四是复用。小函数边界清楚，就能在不同管道里重新组合。今天用于订单统计，明天用于报表导出，不需要复制一份差不多的循环。
## 什么时候不要过度使用函数式写法
函数式编程很好用，但过度使用会把 Python 写得不像 Python。下面几种情况，普通写法通常更合适。
- `lambda` 里出现多层条件判断，改成命名函数。
- `reduce` 的累积逻辑需要读半天，改成 `for` 循环。
- 为了组合而组合，导致调试时看不到中间变量。
- 明明需要维护一组有生命周期的状态，却强行拆成很多无状态函数。
- 团队成员普遍不熟悉柯里化，却把核心业务写成连续调用。
一个实用原则：如果函数式写法让数据流更清楚，就用；如果它只是让代码显得“高级”，就停一下。
## 落到 Python 项目里怎么用
在 Python 项目里，函数式编程更像一套写代码的习惯，而不是一套必须遵守的规矩。纯函数让逻辑更稳定，不可变思路减少意外修改；`map`、`filter`、`reduce`、`sorted` 适合表达数据转换；列表推导式和生成器让代码更贴近 Python；装饰器、`partial`、组合和记忆化则适合处理复用和性能问题。
保留循环，保留类，也保留清晰的中间变量。真正有价值的函数式写法，是让下一位读代码的人更快明白：数据从哪里来，经过了什么规则，最后变成了什么。标签
Python
Python 是一种动态类型、多用途的编程语言。它旨在快速学习、理解和使用，并强制执行干净且统一的语法。
服务端6月19日 12:15
Python 元编程怎么用于框架开发和数据校验？## 先弄清楚：元编程到底在改什么
写 Python 框架时，经常会遇到一种需求：用户只写几行类定义，框架却能自动生成字段、校验规则、查询语句或接口对象。Django ORM、Pydantic、SQLAlchemy、表单库和很多 API SDK 都有这种味道。它们背后常用的就是元编程。
元编程的重点，是把“对象怎么创建、类怎么创建、属性怎么访问”这些过程开放出来。普通业务代码通常操作对象；元编程会再往前一步，操作类、方法、属性协议，甚至在运行时生成类。
常见工具包括：元类、动态属性、动态方法、描述符、`property`、`type()` 动态建类和类装饰器。它们适合做框架层能力，不太适合塞进每个业务函数里。
## 元类：控制类创建过程
在 Python 里，类本身也是对象。普通对象由类创建，类由元类创建。默认情况下，大部分类的元类都是 `type`。
```python
class User:
pass
u = User()
print(type(u)) # <class '__main__.User'>
print(type(User)) # <class 'type'>
```
如果要在“类被创建时”统一检查或改造类，就可以写自定义元类。比如下面这个元类要求类必须声明 `required_attr`，并给类补上一个标记属性：
```python
class ValidateMeta(type):
def __new__(mcls, name, bases, namespace):
if name != 'Base' and 'required_attr' not in namespace:
raise TypeError(f'{name} must define required_attr')
namespace['created_by_meta'] = True
return super().__new__(mcls, name, bases, namespace)
class Base(metaclass=ValidateMeta):
pass
class Service(Base):
required_attr = 'ok'
print(Service.created_by_meta) # True
```
元类常见用途有三类：
- **收集声明式字段**：ORM、表单、序列化框架会扫描类属性，把字段整理到 `_fields` 里。
- **约束类定义**：要求子类必须声明某些属性或方法，或者必须带类型标注。
- **改变实例化行为**：例如注册表、插件系统、单例缓存等。
单例也能用元类实现，但要谨慎。下面的写法会让同一个类只创建一个实例：
```python
class SingletonMeta(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class Config(metaclass=SingletonMeta):
def __init__(self, value):
self.value = value
c1 = Config('first')
c2 = Config('second')
print(c1 is c2) # True
print(c1.value) # second
```
注意这里第二次初始化仍然会执行 `__init__`，所以 `value` 被改成了 `second`。这类细节很容易被忽略，也是元类代码需要写清楚文档的原因。
## 动态属性和动态方法：让对象按规则响应
`__getattr__` 会在正常属性查找失败后触发，适合做“按命名规则生成属性”的能力。`__setattr__` 会拦截赋值，适合把外部赋值统一存进内部字典。
```python
class DynamicRecord:
def __init__(self):
super().__setattr__('_data', {})
def __setattr__(self, name, value):
if name.startswith('_'):
super().__setattr__(name, value)
else:
self._data[name] = value
def __getattr__(self, name):
if name.startswith('get_'):
key = name[4:]
return self._data.get(key)
raise AttributeError(name)
record = DynamicRecord()
record.name = 'Alice'
record.age = 25
print(record.get_name) # Alice
print(record.get_age) # 25
```
动态方法可以直接挂到类上，也可以用 `types.MethodType` 绑定到某个实例上。区别是：挂到类上会成为所有实例的方法，绑定到实例上只影响当前对象。
```python
import types
class Greeter:
pass
def hello(self, name):
return f'Hello, {name}'
Greeter.hello = hello
print(Greeter().hello('Alice')) # Hello, Alice
def only_this(self):
return 'only this instance'
g = Greeter()
g.only_this = types.MethodType(only_this, g)
print(g.only_this())
```
这里最好少碰 `__getattribute__`，因为它会拦截所有属性访问，包括内部属性。除非非常确定查找顺序和递归边界，否则调试成本会很高。
## 描述符：把属性访问做成可复用规则
只要一个对象实现了 `__get__`、`__set__` 或 `__delete__`，它就可以作为描述符使用。描述符最适合做字段校验、懒加载、缓存属性、权限控制这类“每个字段都要同一套规则”的事情。
```python
class ValidatedField:
def __init__(self, expected_type, required=False, default=None):
self.expected_type = expected_type
self.required = required
self.default = default
self.name = None
def __set_name__(self, owner, name):
self.name = '_' + name
def __get__(self, instance, owner):
if instance is None:
return self
return getattr(instance, self.name, self.default)
def __set__(self, instance, value):
if self.required and value is None:
raise ValueError(f'{self.name} is required')
if value is not None and not isinstance(value, self.expected_type):
raise TypeError(f'{self.name} expects {self.expected_type.__name__}')
setattr(instance, self.name, value)
class User:
name = ValidatedField(str, required=True)
age = ValidatedField(int, default=18)
u = User()
u.name = 'Alice'
u.age = 25
print(u.name, u.age)
```
`property` 其实也是描述符的一种常用封装。它更轻，适合单个类里的计算属性或受控赋值；如果同一套字段规则要在很多类里复用，描述符更合适。
```python
class Circle:
def __init__(self, radius):
self.radius = radius
@property
def radius(self):
return self._radius
@radius.setter
def radius(self, value):
if value <= 0:
raise ValueError('radius must be positive')
self._radius = value
@property
def area(self):
return 3.14159 * self._radius ** 2
circle = Circle(5)
print(circle.area)
```
## 用 type 动态创建类
`type(name, bases, namespace)` 可以在运行时创建类。它适合处理“类结构来自配置、数据库、接口 schema”的场景，比如根据 API schema 生成响应对象。
```python
def __init__(self, name):
self.name = name
def greet(self):
return f'Hello, {self.name}'
DynamicUser = type(
'DynamicUser',
(object,),
{
'__init__': __init__,
'greet': greet,
'source': 'runtime'
}
)
user = DynamicUser('Alice')
print(user.greet())
print(user.source)
```
动态类也能创建子类：
```python
class BasePlugin:
def run(self):
return 'base'
Plugin = type('Plugin', (BasePlugin,), {
'name': 'csv_importer',
'run': lambda self: 'import csv'
})
print(Plugin().run()) # import csv
```
这类代码要注意可读性。动态生成的类最好保留清晰的类名、模块名和文档字符串，否则日志、报错和调试器里只会出现一堆看不懂的运行时对象。
## 类装饰器：很多时候比元类更顺手
类装饰器接收一个类，修改后再返回。它不改变类创建机制，理解成本比元类低。只是给类补方法、补属性或做注册时，优先考虑类装饰器。
```python
registry = {}
def register(name):
def decorator(cls):
registry[name] = cls
cls.plugin_name = name
return cls
return decorator
@register('email')
class EmailPlugin:
pass
print(registry['email'] is EmailPlugin)
print(EmailPlugin.plugin_name)
```
类装饰器也能做轻量单例：
```python
def singleton(cls):
instance = None
def wrapper(*args, **kwargs):
nonlocal instance
if instance is None:
instance = cls(*args, **kwargs)
return instance
return wrapper
@singleton
class Database:
pass
```
不过这种写法会把类替换成函数，类型检查、继承和调试信息可能受影响。生产代码里如果要保留完整类语义，通常会选择更明确的工厂函数或容器管理。
## ORM、表单和 API 校验里的典型写法
ORM 是元编程最常见的应用之一。用户声明字段，元类收集字段，描述符控制赋值，模型基类负责初始化。这就是很多框架“少写配置”的来源。
```python
class Field:
def __init__(self, field_type, primary_key=False):
self.field_type = field_type
self.primary_key = primary_key
self.name = None
def __set_name__(self, owner, name):
self.name = name
def __get__(self, instance, owner):
if instance is None:
return self
return instance.__dict__.get(self.name)
def __set__(self, instance, value):
if not isinstance(value, self.field_type):
raise TypeError(f'{self.name} expects {self.field_type.__name__}')
instance.__dict__[self.name] = value
class ModelMeta(type):
def __new__(mcls, name, bases, namespace):
fields = {
key: value
for key, value in namespace.items()
if isinstance(value, Field)
}
namespace['_fields'] = fields
return super().__new__(mcls, name, bases, namespace)
class Model(metaclass=ModelMeta):
def __init__(self, **kwargs):
for name in self._fields:
if name in kwargs:
setattr(self, name, kwargs[name])
def to_dict(self):
return {name: getattr(self, name) for name in self._fields}
class User(Model):
id = Field(int, primary_key=True)
name = Field(str)
age = Field(int)
user = User(id=1, name='Alice', age=25)
print(user.to_dict())
```
表单校验也很像 ORM，只是字段目标从“数据库列”变成了“用户输入”。字段对象负责校验，表单基类负责按字段列表处理输入。
```python
class FormField:
def __init__(self, field_type, required=False, default=None):
self.field_type = field_type
self.required = required
self.default = default
self.name = None
def __set_name__(self, owner, name):
self.name = name
def validate(self, value):
if self.required and value is None:
raise ValueError(f'{self.name} is required')
if value is not None and not isinstance(value, self.field_type):
raise TypeError(f'{self.name} has invalid type')
class FormMeta(type):
def __new__(mcls, name, bases, namespace):
namespace['_fields'] = {
key: value for key, value in namespace.items()
if isinstance(value, FormField)
}
return super().__new__(mcls, name, bases, namespace)
class Form(metaclass=FormMeta):
def __init__(self, **data):
for name, field in self._fields.items():
value = data.get(name, field.default)
field.validate(value)
setattr(self, name, value)
class UserForm(Form):
name = FormField(str, required=True)
age = FormField(int, default=18)
form = UserForm(name='Alice')
print(form.name, form.age)
```
API 响应校验可以用类装饰器来做。接口 schema 变化时，装饰器负责注入 `__init__` 和 `validate`，调用方仍然使用普通类。
```python
def validated_response(schema):
def decorator(cls):
def validate(self, data):
for field, field_type in schema.items():
if field not in data:
raise ValueError(f'missing field: {field}')
if not isinstance(data[field], field_type):
raise TypeError(f'{field} has invalid type')
def __init__(self, data):
self.validate(data)
for key, value in data.items():
setattr(self, key, value)
cls.validate = validate
cls.__init__ = __init__
return cls
return decorator
@validated_response({'name': str, 'age': int, 'email': str})
class UserResponse:
pass
resp = UserResponse({'name': 'Alice', 'age': 25, 'email': 'a@example.com'})
print(resp.email)
```
## 选择哪种元编程工具
可以用一个简单判断来选：
- 只想控制单个属性的读取或赋值，用 `property`。
- 同一套属性规则要复用到多个类，用描述符。
- 想根据字段声明收集类信息，用元类。
- 只想给类补方法、补属性或做注册，用类装饰器。
- 类结构来自运行时配置，再考虑 `type()` 动态建类。
- 只是在对象上临时挂一个方法，用 `types.MethodType` 或直接给类赋函数。
元类能力最强，但通常不是第一选择。很多需求用描述符或类装饰器已经足够，而且更容易被同事读懂。
## 注意事项：性能、调试和文档
元编程会把一部分逻辑藏到类创建、属性访问或装饰阶段。代码能少写，但排查问题时要多看一层。
**性能上**，描述符、`property`、`__getattr__` 都会参与属性访问路径。单次开销通常不大，但在高频循环、序列化大量对象、ORM 批量构造模型时可能放大。昂贵计算可以用缓存描述符：
```python
class cached_property:
def __init__(self, func):
self.func = func
self.name = func.__name__
def __get__(self, instance, owner):
if instance is None:
return self
if self.name not in instance.__dict__:
instance.__dict__[self.name] = self.func(instance)
return instance.__dict__[self.name]
```
**调试上**，动态生成的方法和类要保留可读名称。必要时设置 `__name__`、`__qualname__`、`__module__` 和 `__doc__`，不要让错误栈只剩 `wrapper` 或 `DynamicClass`。
**文档上**，需要把隐式规则写明白：哪些属性会自动生成，什么时候触发校验，元类会给类添加哪些字段，装饰器会不会替换原类。元编程本身没错，麻烦通常来自规则藏得太深。
## 最后怎么取舍
Python 元编程适合放在框架边界：ORM 字段声明、表单校验、API 响应模型、插件注册、缓存属性和运行时类生成。它能把重复样板代码收起来，让使用者写更少的声明。
业务代码里则要克制。能用普通函数解决的，不必上元类；能用类装饰器解决的，不必改类创建流程；能用 `property` 写清楚的，不必做复杂描述符。好的元编程代码应该让调用方更简单，而不是让维护者猜半天。服务端6月19日 10:46
Python 性能优化应该从哪里下手才有效？## 先确认：慢在哪里，别先改代码
Python 性能优化最容易走偏的地方，是一上来就把 `for` 循环改成列表推导式，或者把所有地方都加缓存。真正有效的顺序应该反过来：先测量，再定位瓶颈，最后只改最值得改的那一小段。
一个实用判断是：如果你说不清某段代码慢了多少、占总耗时多少、内存峰值在哪里，那现在还不是优化的时候。先把基准数据拿到手。
### 用 timeit 做小片段基准测试
`timeit` 适合比较一小段代码的耗时，比如列表查找和集合查找、字符串拼接和 `join`。它会重复执行代码，减少单次测量的抖动。
```python
import timeit
setup = "data = list(range(10000)); target = 9999"
list_time = timeit.timeit("target in data", setup=setup, number=10000)
setup = "data = set(range(10000)); target = 9999"
set_time = timeit.timeit("target in data", setup=setup, number=10000)
print(list_time, set_time)
```
注意不要把 `@timeit.timeit` 当装饰器用，`timeit.timeit()` 返回的是耗时数字，不是函数包装器。函数级别的简单计时可以自己写装饰器，或者用 `time.perf_counter()`：
```python
import time
from functools import wraps
def timed(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
cost = time.perf_counter() - start
print(f"{func.__name__}: {cost:.4f}s")
return wrapper
```
### 用 cProfile 看整段程序的热点
`cProfile` 适合回答“到底哪个函数占了最多时间”。它给出的 `cumtime` 很有用，表示函数及其子调用的累计耗时。
```python
import cProfile
import pstats
def main():
data = range(1_000_000)
sum(x * x for x in data)
cProfile.run("main()", "profile.stats")
stats = pstats.Stats("profile.stats")
stats.strip_dirs().sort_stats("cumtime").print_stats(20)
```
如果是线上服务，别只在本地跑一个玩具输入。可以在压测环境里收集真实请求路径，必要时用 `py-spy` 这类采样工具，避免为了分析性能反而把服务拖慢。
### 内存和逐行分析也要看
有些慢不是 CPU 慢，而是内存涨得太快、频繁触发 GC、或者一次性加载了太多数据。可以用 `memory_profiler` 看函数执行期间的内存变化：
```python
# pip install memory-profiler
from memory_profiler import profile
@profile
def load_data():
data = [i for i in range(1_000_000)]
return sum(data)
load_data()
```
`line_profiler` 更适合分析“这个函数里到底哪一行慢”：
```python
# pip install line_profiler
from line_profiler import LineProfiler
def build_result():
result = []
for i in range(10000):
result.append(i * 2)
return sum(result)
lp = LineProfiler()
lp_wrapper = lp(build_result)
lp_wrapper()
lp.print_stats()
```
先用 `cProfile` 找函数，再用 `line_profiler` 看函数内部，通常比一开始就逐行分析更省时间。
## 算法比语法糖更值钱
Python 单行写法再漂亮，也救不了错误的复杂度。一个 O(n²) 的逻辑，在数据量小的时候看不出来，数据一大就会突然变成事故现场。
### 用 set 或 dict 降低查找成本
找重复元素时，双重循环是典型的慢写法：
```python
# O(n²)
def find_duplicates_slow(items):
duplicates = []
for i in range(len(items)):
for j in range(i + 1, len(items)):
if items[i] == items[j] and items[i] not in duplicates:
duplicates.append(items[i])
return duplicates
```
用集合记录状态，复杂度可以降到 O(n)：
```python
# O(n)
def find_duplicates_fast(items):
seen = set()
duplicates = set()
for item in items:
if item in seen:
duplicates.add(item)
else:
seen.add(item)
return list(duplicates)
```
这种优化比把变量名缩短、把循环换成推导式有效得多。
### 减少重复计算
两两计算距离时，如果 `(i, j)` 和 `(j, i)` 都算一遍，工作量直接翻倍：
```python
def calculate_distances(points):
distances = []
for i in range(len(points)):
for j in range(i + 1, len(points)):
dx = points[j][0] - points[i][0]
dy = points[j][1] - points[i][1]
distances.append((dx * dx + dy * dy) ** 0.5)
return distances
```
还能继续优化吗？要看业务是否真的需要保存所有距离。如果只是找最小值，就不必把中间结果全部放进列表。
## 数据结构会直接影响速度和内存
Python 的列表、集合、字典、元组都很好用，但适用场景不一样。性能问题经常不是“Python 不够快”，而是数据结构选错了。
### 列表适合顺序遍历，集合适合成员判断
```python
users = list(range(100000))
user_set = set(users)
99999 in users # 需要顺序扫描
99999 in user_set # 哈希查找，平均 O(1)
```
如果一段代码每天要做几百万次成员判断，把列表换成集合通常很划算。但集合会占用更多内存，也不保留重复元素，不能只看速度。
### 生成器适合流式处理
一次性构造大列表，会把所有结果放进内存。只需要逐个消费时，用生成器更稳。
```python
def squares(n):
for i in range(n):
yield i * i
total = sum(squares(1_000_000))
```
列表推导式不是坏东西。需要多次遍历、随机访问、切片时，列表更方便。生成器的优势在于“边算边用”，不是所有场景都要换成生成器。
### __slots__ 能省内存，但会牺牲灵活性
当你需要创建大量结构相同的小对象时，`__slots__` 可以减少每个对象的 `__dict__` 开销。
```python
class Point:
__slots__ = ("x", "y")
def __init__(self, x, y):
self.x = x
self.y = y
```
它的代价是不能随意添加新属性，和某些依赖 `__dict__` 的库也可能不兼容。几十个对象没必要上 `__slots__`，几十万、几百万个对象才值得认真比较。
## 多用内置函数，但别迷信“一行代码”
`sum()`、`max()`、`min()`、`sorted()`、`any()`、`all()` 这类内置函数通常跑得更快，因为核心循环在 C 层实现。
```python
def manual_sum(items):
total = 0
for item in items:
total += item
return total
builtin_sum = sum
```
不过可读性也要算账。为了把三段清晰的业务逻辑压成一个嵌套表达式，最后没人敢改，这不是优化，是埋雷。
## I/O 优化先看批量、缓冲和异步
文件、网络、数据库这些 I/O 场景，瓶颈通常不在 Python 运算本身，而在等待外部系统响应。优化方向也不一样。
### 文件读写尽量批量处理
逐行写入不是不能用，但如果数据已经在内存里，批量写入会少很多系统调用。
```python
def write_lines_fast(filename, lines):
with open(filename, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
```
读取大文件时，不要为了“省事”直接 `read()` 全部加载：
```python
def process_large_file(filename):
with open(filename, "r", encoding="utf-8") as f:
for line in f:
yield line.strip()
```
缓冲大小可以调，但别在文本模式下写 `buffering=0`，无缓冲只允许二进制模式。大多数情况下，默认缓冲已经够用。
### 网络 I/O 适合异步或线程池
如果任务主要在等网络返回，异步 I/O 或线程池都能提高吞吐。下面是 `aiohttp` 的常见写法：
```python
import asyncio
import aiohttp
async def fetch(session, url):
async with session.get(url, timeout=10) as response:
response.raise_for_status()
return await response.text()
async def fetch_all(urls):
async with aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
return await asyncio.gather(*tasks)
```
真实项目里还要限制并发数，不要同时打出几千个请求把对方服务或自己机器打满。
## 并发要分清 CPU 密集还是 I/O 密集
Python 有 GIL，线程不适合加速纯 CPU 计算，但很适合等待网络、磁盘、数据库这类 I/O。CPU 密集任务通常考虑多进程、C 扩展、NumPy、Cython，或者把计算交给更合适的服务。
### I/O 密集用 ThreadPoolExecutor
```python
from concurrent.futures import ThreadPoolExecutor
import requests
def download(url):
response = requests.get(url, timeout=10)
response.raise_for_status()
return len(response.content)
with ThreadPoolExecutor(max_workers=8) as executor:
sizes = list(executor.map(download, urls))
```
### CPU 密集用 ProcessPoolExecutor
```python
from concurrent.futures import ProcessPoolExecutor
def compute(chunk):
return sum(x * x for x in chunk)
with ProcessPoolExecutor(max_workers=4) as executor:
result = sum(executor.map(compute, chunks))
```
多进程有序列化和进程通信成本。数据块太小、任务太轻时，开进程可能比单进程更慢。
## 缓存能救热点，也能制造脏数据
缓存适合重复计算、重复查询、重复读取的场景。它不适合掩盖慢 SQL，也不适合缓存所有东西。
### lru_cache 适合纯函数
```python
from functools import lru_cache
@lru_cache(maxsize=1024)
def get_user_permissions(user_id):
return load_permissions_from_db(user_id)
```
`lru_cache` 的参数必须可哈希；结果最好由参数唯一决定。如果函数依赖当前时间、登录态、外部可变配置，就要谨慎。
### Redis 适合跨进程缓存
```python
import json
import redis
r = redis.Redis(host="localhost", port=6379, db=0)
def set_cache(key, value, ttl=300):
r.setex(key, ttl, json.dumps(value, ensure_ascii=False))
def get_cache(key):
raw = r.get(key)
return json.loads(raw) if raw else None
```
示例里用 JSON 是为了避免随手 `pickle.loads()` 带来的安全风险。只有在数据来源完全可信时，才考虑 pickle。
缓存一定要设计失效策略。TTL 太长会读到旧数据，太短又没有命中率；热点 key 还要考虑击穿、雪崩和并发回源。
## 字符串和对象创建也有细节
大量字符串拼接时，`join()` 通常比循环里的 `+` 更合适：
```python
def build_text(parts):
return "".join(parts)
```
格式化字符串优先用 f-string，清晰也快：
```python
message = f"Name: {name}, Age: {age}"
```
处理大小写、替换、分割时，先看字符串内置方法。手写循环不仅慢，也更容易漏掉边界情况。
## 数据库优化别只盯 Python 代码
很多接口慢，Python 只背了一半锅，另一半在数据库。
### 连接池减少连接开销
```python
from sqlalchemy import create_engine
engine = create_engine(
"postgresql://user:password@localhost/dbname",
pool_size=10,
max_overflow=5,
pool_pre_ping=True,
)
```
连接池不是越大越好。池子太大会把数据库连接打满，反而拖垮整体服务。
### 批量写入比逐条写入稳定
```python
def insert_fast(cursor, items):
cursor.executemany(
"INSERT INTO events(name, value) VALUES (%s, %s)",
[(item.name, item.value) for item in items],
)
```
ORM 也通常提供 bulk insert，但要确认它是否跳过了模型钩子、默认值、校验逻辑。
### 索引要服务查询条件
```sql
CREATE INDEX idx_users_name ON users(name);
SELECT * FROM users WHERE name = 'Alice';
```
如果查询写成 `WHERE LOWER(name) = 'alice'`，普通 `name` 索引可能用不上。要么调整查询方式，要么建立对应的函数索引。遇到慢查询，先看 `EXPLAIN`，不要凭感觉加索引。
## 监控决定优化有没有持续价值
本地优化通过之后，还要看线上是否真的变快。至少记录这些指标：接口耗时、P95/P99 延迟、错误率、内存峰值、队列长度、缓存命中率、数据库慢查询。
一个简单的耗时日志装饰器可以这样写：
```python
import logging
import time
from functools import wraps
logger = logging.getLogger(__name__)
def log_cost(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
return func(*args, **kwargs)
finally:
cost = time.perf_counter() - start
logger.info("%s cost %.4fs", func.__name__, cost)
return wrapper
```
更完整的项目会把指标打到 Prometheus、StatsD、OpenTelemetry 或 APM 系统里。单次平均值不够，P95/P99 更能反映用户真实体感。
## 优化时必须保留的取舍
性能优化不是越快越好，至少要同时看四件事：
- **可读性**：快 3%，但代码复杂一倍，通常不值得。
- **正确性**：缓存、并发、批量写入都会引入一致性和边界问题。
- **内存**：用空间换时间很常见，但容器内存有限。
- **可观测性**：没有监控，优化效果很快会被后续改动吃掉。
如果只能记住一个顺序，那就是：先用 `timeit`、`cProfile`、内存分析和线上指标确认瓶颈；再从算法、数据结构、I/O、并发、缓存、数据库这些高收益位置下手；最后用同样的指标复测。Python 性能优化最怕“感觉上更快”，真正可靠的是前后两组数据。服务端6月19日 10:46
Python 深拷贝和浅拷贝有什么区别，什么时候该用？## 先把“赋值”和“拷贝”分开
很多 Python 拷贝问题，最容易错在第一步：把赋值当成了复制。
赋值只是多了一个名字，两个变量仍然指向同一个对象。拷贝才会创建一个新对象。至于是只复制外层，还是连里面的对象一起复制，这就是浅拷贝和深拷贝的区别。
```python
original = [1, 2, 3]
assigned = original
assigned[0] = 99
print(original) # [99, 2, 3]
```
上面没有发生拷贝，`assigned` 和 `original` 指向同一个列表。
如果使用 `copy.copy()`，外层列表会变成两个对象：
```python
import copy
original = [1, 2, 3]
copied = copy.copy(original)
copied[0] = 99
print(original) # [1, 2, 3]
```
这段代码看起来已经解决问题了，但只适用于没有嵌套可变对象的情况。真正让人踩坑的，通常是列表里还有列表、字典里还有列表、对象属性里还有可变对象。
## 浅拷贝复制外层，里面的对象仍然共享
浅拷贝会创建一个新的“外壳”。外层容器是新的，但容器里的元素还是原来的引用。
```python
import copy
original = [1, 2, [3, 4]]
shallow = copy.copy(original)
print(original is shallow) # False，外层列表不同
print(original[2] is shallow[2]) # True，嵌套列表仍然是同一个
```
所以修改顶层元素没问题：
```python
shallow[0] = 99
print(original) # [1, 2, [3, 4]]
```
但修改嵌套对象，就会影响原数据：
```python
shallow[2][0] = 99
print(original) # [1, 2, [99, 4]]
```
这也是很多“明明复制了一份，原数据还是被改了”的来源。
## 常见浅拷贝写法有哪些
不同容器有自己的浅拷贝写法。它们写法不同，本质接近：只复制外层容器。
```python
import copy
items = [1, 2, [3, 4]]
copy1 = copy.copy(items)
copy2 = items.copy()
copy3 = items[:]
copy4 = list(items)
```
字典也类似：
```python
original = {"debug": False, "endpoints": ["api1", "api2"]}
cloned = original.copy()
cloned["debug"] = True
cloned["endpoints"].append("api3")
print(original["debug"]) # False
print(original["endpoints"]) # ['api1', 'api2', 'api3']
```
`debug` 是顶层值，改副本不会影响原字典；`endpoints` 是嵌套列表，浅拷贝后仍然共享。
## 深拷贝会递归复制嵌套对象
深拷贝使用 `copy.deepcopy()`。它会尽量把对象内部引用到的对象也复制一份。
```python
import copy
original = [1, 2, [3, 4]]
deep = copy.deepcopy(original)
deep[2][0] = 99
print(original) # [1, 2, [3, 4]]
```
再看一个更接近业务数据的例子：
```python
import copy
data = {
"users": [
{"name": "Alice", "scores": [85, 90, 78]},
{"name": "Bob", "scores": [92, 88, 95]},
]
}
processed = copy.deepcopy(data)
for user in processed["users"]:
user["average"] = sum(user["scores"]) / len(user["scores"])
print("average" in data["users"][0]) # False
```
如果这里只做浅拷贝，`users` 列表和里面的用户字典仍然可能被共享，后续处理就容易污染原始数据。
## 可变对象和不可变对象对拷贝结果影响很大
讨论深浅拷贝时，先看对象能不能被原地修改。
- 不可变对象：`int`、`float`、`str`、`tuple`、`frozenset` 等。
- 可变对象：`list`、`dict`、`set`、大多数自定义对象等。
不可变对象通常没必要复制。对整数、字符串这类对象，`copy.copy()` 和 `copy.deepcopy()` 往往会返回原对象。
```python
import copy
a = 42
b = copy.copy(a)
c = copy.deepcopy(a)
print(a is b) # True
print(a is c) # True
```
但元组要稍微小心。元组本身不可变，不代表它里面的元素都不可变。
```python
import copy
t = (1, 2, [3, 4])
shallow = copy.copy(t)
deep = copy.deepcopy(t)
print(t is shallow) # True，浅拷贝直接复用元组
print(t[2] is deep[2]) # False，深拷贝复制了里面的列表
```
如果一个不可变容器里藏着可变对象，深拷贝仍然有意义。
## 什么时候用浅拷贝，什么时候用深拷贝
可以按“后面会不会改嵌套对象”来判断。
| 场景 | 更合适的方式 | 原因 |
| --- | --- | --- |
| 只改列表或字典的顶层元素 | 浅拷贝 | 外层独立就够了，成本低 |
| 要改嵌套列表、嵌套字典 | 深拷贝 | 避免共享内部对象 |
| 数据完全只读 | 不拷贝 | 拷贝只会增加开销 |
| 配置模板生成多份独立配置 | 深拷贝 | 每份配置可能修改嵌套项 |
| 缓存对外返回数据 | 通常深拷贝 | 防止调用方改坏缓存 |
| 大型数据结构且性能敏感 | 谨慎深拷贝 | 可能复制大量对象 |
浅拷贝不是“低级版深拷贝”。它只是适合不同场景。比如复制一个简单列表后只追加顶层元素，用浅拷贝就很好：
```python
items = [1, 2, 3]
new_items = items.copy()
new_items.append(4)
print(items) # [1, 2, 3]
```
## 配置、数据处理、缓存、撤销功能里的用法
### 配置模板
配置对象经常有嵌套字段，比如 endpoints、headers、feature flags。用默认配置派生新配置时，深拷贝更稳妥。
```python
import copy
default_config = {
"debug": False,
"max_retries": 3,
"timeout": 30,
"endpoints": ["api1.example.com", "api2.example.com"],
}
config = copy.deepcopy(default_config)
config["debug"] = True
config["endpoints"].append("api3.example.com")
print(default_config["debug"]) # False
print(default_config["endpoints"]) # ['api1.example.com', 'api2.example.com']
```
### 数据处理
做清洗、补字段、格式转换时，是否拷贝取决于函数约定。如果函数承诺“不修改输入”，要么在内部复制，要么返回新数据结构。
```python
import copy
def add_average(data):
result = copy.deepcopy(data)
for user in result["users"]:
user["average"] = sum(user["scores"]) / len(user["scores"])
return result
```
更推荐把这个约定写清楚：函数会不会修改参数，比它内部用了什么拷贝方式更重要。
### 缓存数据
缓存最怕外部代码拿到引用后直接修改。
```python
import copy
class DataCache:
def __init__(self):
self._cache = {}
def set(self, key, value):
self._cache[key] = copy.deepcopy(value)
def get(self, key):
if key not in self._cache:
return None
return copy.deepcopy(self._cache[key])
cache = DataCache()
cache.set("data", {"items": [1, 2, 3]})
cached = cache.get("data")
cached["items"].append(4)
print(cache.get("data")) # {'items': [1, 2, 3]}
```
这里 `set` 和 `get` 都做深拷贝，是为了让缓存内部状态和调用方彻底隔开。代价是性能会下降，数据很大时要评估是否值得。
### 撤销和重做
撤销功能保存的是“当时的状态”。如果状态里有嵌套对象，通常要保存深拷贝快照。
```python
import copy
class EditorState:
def __init__(self):
self.document = {"blocks": []}
self.history = []
def save(self):
self.history.append(copy.deepcopy(self.document))
def add_block(self, text):
self.save()
self.document["blocks"].append({"text": text})
def undo(self):
if self.history:
self.document = self.history.pop()
```
如果只保存浅拷贝，历史记录里的嵌套 block 可能继续被当前文档共享，撤销出来的状态就不可靠。
## 自定义对象如何控制拷贝行为
对自定义类，Python 会尝试按对象属性进行拷贝。但有些对象需要自己定义规则：比如连接池、文件句柄、缓存、父子节点引用等。
可以实现 `__copy__` 和 `__deepcopy__`。
```python
import copy
class Settings:
def __init__(self, name, options):
self.name = name
self.options = options
def __copy__(self):
cls = type(self)
new_obj = cls(self.name, self.options)
return new_obj
def __deepcopy__(self, memo):
cls = type(self)
new_obj = cls(
copy.deepcopy(self.name, memo),
copy.deepcopy(self.options, memo),
)
memo[id(self)] = new_obj
return new_obj
original = Settings("prod", {"endpoints": ["api1"]})
shallow = copy.copy(original)
deep = copy.deepcopy(original)
shallow.options["endpoints"].append("api2")
print(original.options) # {'endpoints': ['api1', 'api2']}
deep.options["endpoints"].append("api3")
print(original.options) # {'endpoints': ['api1', 'api2']}
```
`__deepcopy__(self, memo)` 里的 `memo` 很关键。它用来记录“某个对象已经复制过了”，既能保持共享关系，也能处理循环引用。
## 循环引用为什么不会轻易拖垮 deepcopy
循环引用就是对象之间互相指向。比如链表节点指回前一个节点，树节点存 parent，图结构里节点互相关联。
```python
import copy
a = [1, 2]
b = [3, 4]
a.append(b)
b.append(a)
cloned = copy.deepcopy(a)
print(cloned[2][2] is cloned) # True
```
`deepcopy` 能处理这个例子，是因为它内部使用了 `memo` 字典。复制对象前先查 `memo`，如果这个对象已经复制过，就直接复用那份副本，而不是无限递归下去。
自定义 `__deepcopy__` 时也要配合 `memo`：
```python
import copy
class Node:
def __init__(self, value):
self.value = value
self.next = None
def __deepcopy__(self, memo):
if id(self) in memo:
return memo[id(self)]
new_node = type(self)(self.value)
memo[id(self)] = new_node
new_node.next = copy.deepcopy(self.next, memo)
return new_node
node1 = Node(1)
node2 = Node(2)
node1.next = node2
node2.next = node1
copied = copy.deepcopy(node1)
print(copied.next.next is copied) # True
```
如果忘了写入 `memo`，遇到循环结构就可能递归到报错。
## 性能上别把 deepcopy 当默认选项
深拷贝要遍历对象图，数据越大、嵌套越深、对象越复杂，成本越高。它不只是“慢一点”，有时会复制出大量本来不需要复制的对象。
可以用一个简单脚本感受差异：
```python
import copy
import time
large_data = {
"items": [{"id": i, "tags": ["python", "copy"]} for i in range(10000)]
}
start = time.perf_counter()
shallow = copy.copy(large_data)
print(f"shallow: {time.perf_counter() - start:.6f}s")
start = time.perf_counter()
deep = copy.deepcopy(large_data)
print(f"deep: {time.perf_counter() - start:.6f}s")
```
实际项目里，更常见的优化不是“让 deepcopy 更快”，而是减少不必要的复制：
- 只读数据直接传引用。
- 只改顶层结构时用浅拷贝。
- 只需要修改某个分支时，手动复制那条路径。
- 大对象考虑不可变数据结构，或者明确函数的修改边界。
- 缓存场景权衡安全和性能，不要对超大结果无脑深拷贝。
例如只改一个用户的分数，不一定要复制整份数据：
```python
data = {
"users": {
"alice": {"scores": [85, 90]},
"bob": {"scores": [92, 88]},
}
}
new_data = data.copy()
new_data["users"] = data["users"].copy()
new_data["users"]["alice"] = data["users"]["alice"].copy()
new_data["users"]["alice"]["scores"] = data["users"]["alice"]["scores"].copy()
new_data["users"]["alice"]["scores"].append(100)
print(data["users"]["alice"]["scores"]) # [85, 90]
```
这段写法比 `deepcopy(data)` 啰嗦，但在大数据结构里可能更可控。
## 容易忽略的边界
### 有些对象不适合拷贝
模块、函数、文件对象、网络连接、数据库连接这类对象，通常不应该被深拷贝。对它们来说，“复制一份”本身就不一定有合理语义。
```python
import copy
f = open("example.txt", "w")
try:
copy.deepcopy(f)
except TypeError as e:
print(type(e).__name__)
finally:
f.close()
```
如果类里持有这类资源，最好在 `__copy__` 或 `__deepcopy__` 里明确处理：共享、重新创建，或者直接禁止复制。
### 深拷贝会保持共享关系
如果原对象里两个字段指向同一个列表，`deepcopy` 后它们仍然会指向同一个“副本列表”，而不是变成两个独立列表。
```python
import copy
shared = [1, 2]
original = {"a": shared, "b": shared}
cloned = copy.deepcopy(original)
print(cloned["a"] is cloned["b"]) # True
print(cloned["a"] is shared) # False
```
这也是 `memo` 的作用之一：不是机械地“见一次复制一次”，而是在副本里保留原来的引用关系。
## 一套实用判断方式
写代码时可以先问三个问题：
1. 后续会不会修改这个对象？不会，就别拷贝。
2. 只改外层，还是会改里面的列表、字典、对象属性？只改外层，用浅拷贝；会改里面，用深拷贝或手动复制相关分支。
3. 数据大不大、对象里有没有资源句柄或循环引用？如果有，别直接把 `deepcopy` 当万能按钮。
一句话概括：赋值不会复制对象；浅拷贝只换外壳；深拷贝会递归复制内部对象，但要付出性能和语义成本。配置模板、数据清洗、缓存隔离、撤销快照这些场景，深拷贝很有用；简单列表、只读数据和性能敏感路径，浅拷贝甚至不拷贝往往更合适。服务端6月4日 15:48
Python上下文管理器：__exit__异常处理、@contextmanager和ExitStack`with open(...) as f:` 这行代码用了十几年，但很多人不知道背后的机制——上下文管理器。更关键的是，自己写一个靠谱的上下文管理器并不简单：`__exit__` 里该不该吃掉异常？`@contextmanager` 的 yield 和 finally 怎么配合？多个资源怎么一起管理？这篇文章把这些问题都讲清楚。
## 上下文管理器解决什么问题
资源管理的核心要求：**用完必须释放，不管有没有异常**。不用上下文管理器就得写 try/finally：
```python
# 笨办法
f = open('data.txt')
try:
data = f.read()
finally:
f.close()
# with 语句：等价但简洁
with open('data.txt') as f:
data = f.read()
# f.close() 自动调用，即使 read() 抛异常
```
`with` 语句保证 `__exit__` 一定被调用，省掉 finally 的样板代码。文件、数据库连接、锁、事务，都需要这种保证。
## __enter__ 和 __exit__ 协议
上下文管理器是实现 `__enter__` 和 `__exit__` 的对象：
```python
class Timer:
def __enter__(self):
import time
self.start = time.perf_counter()
return self # as 绑定的对象
def __exit__(self, exc_type, exc_val, exc_tb):
import time
elapsed = time.perf_counter() - self.start
print(f"耗时: {elapsed:.4f}s")
return False # 不吞异常，继续传播
with Timer() as t:
time.sleep(1)
# 耗时: 1.0012s
```
`__enter__` 返回值通过 `as` 绑定。`__exit__` 的三个参数是异常信息（没有异常时都是 None）。返回 `True` 表示吞掉异常，返回 `False`（或不返回）让异常继续传播。
### __exit__ 该不该吞异常
大多数场景**不应该吞**——返回 False，让调用方处理异常。只有极少数场景需要吞：如 `suppress(FileNotFoundError)` 这种明确要忽略特定异常的。
```python
# 危险：吞掉所有异常，问题难以排查
def __exit__(self, exc_type, exc_val, exc_tb):
return True # ❌ 任何异常都被静默忽略
# 正确：只吞特定异常
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is FileNotFoundError:
return True # 忽略文件不存在
return False # 其他异常继续传播
```
## @contextmanager：更简单的写法
手写 `__enter__` 和 `__exit__` 容易出错——`@contextmanager` 装饰器把上下文管理器变成生成器函数，逻辑更清晰：
```python
from contextlib import contextmanager
@contextmanager
def timer(name="block"):
import time
start = time.perf_counter()
try:
yield # yield 处就是 with 块的代码
finally:
elapsed = time.perf_counter() - start
print(f"{name} 耗时: {elapsed:.4f}s")
with timer("数据处理"):
process_data()
```
`yield` 之前是 `__enter__`，`yield` 之后是 `__exit__`。`finally` 块保证清理逻辑一定执行。
### yield 可以返回值
```python
@contextmanager
def temp_directory():
import tempfile, shutil
dirpath = tempfile.mkdtemp()
try:
yield dirpath # 返回临时目录路径
finally:
shutil.rmtree(dirpath) # 用完删除
with temp_directory() as tmpdir:
print(f"临时目录: {tmpdir}")
# 在 tmpdir 里写文件...
# tmpdir 已被删除
```
### 异常处理
`@contextmanager` 里 `yield` 抛出的异常会传播到 `with` 块，但如果在 yield 外面捕获了，就相当于吞掉：
```python
@contextmanager
def safe_operation():
try:
yield
except ValueError as e:
print(f"捕获到 ValueError: {e}")
# 不 re-raise，异常被吞掉
finally:
print("清理完成")
with safe_operation():
raise ValueError("出错了")
# 输出：捕获到 ValueError: 出错了
# 清理完成
# 程序继续执行，不会崩溃
```
如果想在 `@contextmanager` 里**记录但不吞掉**异常，用 `raise` 重新抛出，或者不捕获让 finally 执行后自然传播。
## contextlib 实用工具
### suppress：忽略指定异常
替代 `try/except` + `pass` 的惯用法：
```python
from contextlib import suppress
# 以前
try:
os.remove('temp.txt')
except FileNotFoundError:
pass
# 现在
with suppress(FileNotFoundError):
os.remove('temp.txt')
```
可以忽略多种异常：`suppress(FileNotFoundError, PermissionError)`。
### closing：给有 close() 方法的对象加 with 支持
```python
from contextlib import closing
from urllib.request import urlopen
with closing(urlopen('https://example.com')) as response:
data = response.read()
# response.close() 自动调用
```
### redirect_stdout/redirect_stderr：临时重定向输出
```python
from contextlib import redirect_stdout
import io
output = io.StringIO()
with redirect_stdout(output):
print("这行不会显示在终端")
captured = output.getvalue() # "这行不会显示在终端
"
```
适合测试里捕获 print 输出，或者把进度信息写到日志文件而不是终端。
### ExitStack：动态管理多个上下文
不确定需要打开多少个资源时用 ExitStack：
```python
from contextlib import ExitStack
files = ['a.txt', 'b.txt', 'c.txt']
with ExitStack() as stack:
handles = [stack.enter_context(open(f)) for f in files]
# 三个文件都打开了，任何一个打开失败，之前打开的会自动关闭
for h in handles:
process(h)
# 三个文件全部自动关闭
```
也可以用 `callback` 注册清理函数：
```python
with ExitStack() as stack:
stack.callback(print, "清理完成")
do_something()
# 无论 do_something 是否抛异常，"清理完成" 都会打印
```
## 异步上下文管理器
Python 3.5+ 支持 `async with`，对应 `__aenter__` 和 `__aexit__`：
```python
class AsyncDBConnection:
async def __aenter__(self):
self.conn = await create_connection()
return self.conn
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.conn.close()
return False
async def main():
async with AsyncDBConnection() as conn:
await conn.execute("SELECT 1")
```
`@asynccontextmanager` 是异步版本：
```python
from contextlib import asynccontextmanager
@asynccontextmanager
async def db_transaction(pool):
conn = await pool.acquire()
try:
yield conn
finally:
await pool.release(conn)
async def main():
async with db_transaction(pool) as conn:
await conn.execute("INSERT ...")
```
FastAPI 和 SQLAlchemy 的异步数据库会话就是用这个模式。
## 实际应用场景
### 数据库事务
```python
@contextmanager
def transaction(conn):
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
with transaction(conn):
conn.execute("INSERT ...")
conn.execute("UPDATE ...")
# 两条语句要么都成功，要么都回滚
```
### 临时修改环境变量
```python
@contextmanager
def env_var(key, value):
import os
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 env_var('DATABASE_URL', 'sqlite:///:memory:'):
run_tests() # 测试时用内存数据库
# DATABASE_URL 恢复原值
```
### 线程锁
`threading.Lock` 本身就是上下文管理器：
```python
import threading
lock = threading.Lock()
with lock:
# 临界区，自动加锁/释放锁
update_shared_data()
```
比 `lock.acquire() ... lock.release()` 安全——不会因为异常导致锁不释放。服务端6月2日 01:38
Python 装饰器是怎么工作的？@ 语法糖、执行时机和 wraps 详解装饰器的本质是一个接收函数作为参数并返回新函数的高阶函数。`@decorator` 语法糖等价于 `func = decorator(func)`。理解装饰器的关键：它只是函数替换，在定义时执行，不是调用时。
## 装饰器做了什么
```python
def log(func):
def wrapper(*args, **kwargs):
print(f"Calling {func.__name__}")
return func(*args, **kwargs)
return wrapper
@log
def hello(name):
print(f"Hello, {name}")
# 等价于
hello = log(hello)
```
`@log` 发生在函数定义时，不是调用时。Python 解释器看到 `@log` 后，把 `hello` 传给 `log()`，返回的 `wrapper` 替换掉原来的 `hello`。之后所有对 `hello()` 的调用实际上调的是 `wrapper()`。
## 执行时机
```python
def log(func):
print("装饰器执行了") # 模块加载时就执行
def wrapper(*args, **kwargs):
print("wrapper 执行了") # 每次调用函数时执行
return func(*args, **kwargs)
return wrapper
@log
def hello():
print("hello")
# 输出: 装饰器执行了（定义时立即执行）
hello()
# 输出: wrapper 执行了 / hello
```
装饰器外层的代码在模块加载时执行（一次），wrapper 内的代码在每次函数调用时执行。
## 多个装饰器的叠加
```python
@a
@b
def f(): pass
# 等价于
f = a(b(f))
```
从下到上装饰。调用 `f()` 时，执行顺序是 a -> b -> f -> b 的后处理 -> a 的后处理。
## 带参数的装饰器
需要三层嵌套——最外层接收装饰器参数，中间层接收被装饰函数，最内层是 wrapper：
```python
def retry(max_attempts=3):
def decorator(func):
def wrapper(*args, **kwargs):
for i in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception:
if i == max_attempts - 1:
raise
return wrapper
return decorator
@retry(max_attempts=5)
def call_api():
...
# 等价于
call_api = retry(max_attempts=5)(call_api)
```
`@retry(5)` 先调用 `retry(5)` 返回 `decorator`，再 `decorator(call_api)` 返回 `wrapper`。三层嵌套是固定模式。
## functools.wraps
装饰器替换了原函数，导致 `__name__`、`__doc__` 等元信息丢失。`@wraps(func)` 复制原函数的元信息到 wrapper：
```python
from functools import wraps
def log(func):
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
@log
def hello(): pass
print(hello.__name__) # hello（没有 @wraps 会是 wrapper）
```
## 装饰器 vs 闭包
装饰器是基于闭包的。`wrapper` 闭包了 `func` 变量——每次调用 `wrapper` 时都能访问到被装饰的原始函数。装饰器就是"创建闭包的工厂函数"。
## 追问
### 装饰器能装饰类吗？
能。装饰器接收的参数不一定是函数，也可以是类：
```python
def add_repr(cls):
def __repr__(self):
return f"{cls.__name__}({self.__dict__})"
cls.__repr__ = __repr__
return cls
@add_repr
class User:
def __init__(self, name):
self.name = name
```
dataclass 的 `@dataclass` 就是类装饰器。
### 装饰器有性能开销吗？
有一层函数调用的开销，通常可以忽略。但在极高性能场景（百万次/秒调用），装饰器的额外调用栈可能成为瓶颈。可以用 `functools.lru_cache` 缓存结果避免重复执行。服务端6月2日 01:37
Python GIL 是什么？为什么多线程不能利用多核？怎么绕过？GIL（Global Interpreter Lock）是 CPython 的一把全局互斥锁，同一时刻只允许一个线程执行 Python 字节码。这意味着 Python 多线程无法利用多核 CPU 做计算密集型任务。但 IO 密集型任务不受影响——线程等待 IO 时会释放 GIL。
## GIL 为什么存在
CPython 的内存管理（引用计数）不是线程安全的。如果多个线程同时修改 `ob_refcnt`，可能导致内存泄漏或提前释放。GIL 是最简单的解决方案——一个线程执行 Python 代码时，其他线程不能运行。
为什么不去掉？Python 核心开发者试过多次，去掉 GIL 会导致单线程性能下降 10-30%（因为要加细粒度锁替代全局锁），C 扩展也需要大量改写。社区不愿意接受这个代价。
Python 3.13 引入了实验性的 free-threaded 模式（PEP 703），允许禁用 GIL，但目前还是实验阶段，性能和兼容性都不成熟。
## GIL 的影响范围
| 任务类型 | 多线程表现 | 原因 |
|----------|-----------|------|
| CPU 密集 | 比单线程还慢 | 线程争抢 GIL，切换开销大 |
| IO 密集 | 有效加速 | 等 IO 时释放 GIL，其他线程可运行 |
| 混合型 | 部分有效 | 计算部分受 GIL 限制 |
CPU 密集比单线程还慢是因为 GIL 切换本身有开销，多线程反而增加了竞争。
## 绕过 GIL 的三种方式
**1. 多进程（最常用）**
```python
from multiprocessing import Pool
def heavy_compute(n):
return sum(i * i for i in range(n))
with Pool(4) as p:
results = p.map(heavy_compute, [10**7] * 4) # 4 个进程并行
```
每个进程有独立的 GIL，真正并行。缺点：进程间通信成本高（数据要序列化），启动慢。
**2. C 扩展释放 GIL**
在 C 扩展中做计算密集型操作时，可以手动释放 GIL：
```c
Py_BEGIN_ALLOW_THREADS
// 这里做纯 C 计算，不操作 Python 对象
result = heavy_computation(data);
Py_END_ALLOW_THREADS
```
NumPy、Pandas、hashlib 等库在底层 C 代码中释放了 GIL，所以它们在多线程中可以并行运行。
**3. 用 asyncio 替代多线程**
IO 密集型任务不需要多线程，用协程就够了：
```python
import asyncio
import aiohttp
async def fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.text()
async def main():
tasks = [fetch(url) for url in urls]
results = await asyncio.gather(*tasks)
```
单线程 + 事件循环，没有 GIL 问题，也没有线程切换开销。
## GIL 的释放时机
CPython 在两种情况下释放 GIL：
1. **IO 操作**：网络请求、文件读写、time.sleep() 等
2. **C 扩展主动释放**：NumPy 运算、hashlib 哈希等
纯 Python 代码（循环、计算、字符串操作）永远不会释放 GIL。
## 追问
### Python 3.13 的 no-GIL 模式能用吗？
实验阶段，不推荐生产使用。性能比有 GIL 模式慢约 10-15%，很多 C 扩展还不兼容。等 Python 3.14/3.15 稳定后再考虑。
### 多线程在 Python 里完全没用吗？
不是。IO 密集型场景（网络爬虫、API 调用、数据库查询）多线程比单线程快得多。只是 CPU 密集型场景多线程没有加速效果。服务端6月2日 01:36
Python 迭代器和生成器有什么区别？yield 和迭代器协议详解迭代器是实现了 `__iter__` 和 `__next__` 方法的对象，生成器是用 `yield` 关键字自动创建的迭代器。生成器是迭代器的子集——所有生成器都是迭代器，但迭代器不一定是生成器。核心区别：生成器更简洁，且天然支持惰性求值。
## 迭代器协议
迭代器必须实现两个方法：
- `__iter__`：返回 self（让迭代器本身也可迭代）
- `__next__`：返回下一个值，没有值时抛出 StopIteration
```python
class Countdown:
def __init__(self, start):
self.current = start
def __iter__(self):
return self
def __next__(self):
if self.current <= 0:
raise StopIteration
self.current -= 1
return self.current + 1
for n in Countdown(3):
print(n) # 3, 2, 1
```
手动实现迭代器要写 `__iter__`、`__next__`、维护状态、处理 StopIteration。代码量多，容易写错。
## 生成器：迭代器的语法糖
生成器用 `yield` 关键字，Python 自动实现迭代器协议：
```python
def countdown(start):
current = start
while current > 0:
yield current
current -= 1
for n in countdown(3):
print(n) # 3, 2, 1
```
调用 `countdown(3)` 不会执行函数体，而是返回一个生成器对象。每次 `next()` 执行到 `yield` 暂停并返回值，下次 `next()` 从暂停处继续。
3 行代码 vs 10 行代码，效果完全一样。这就是为什么 Python 社区几乎总是用生成器而不是手写迭代器。
## 生成器表达式
类似列表推导式，但用圆括号，惰性求值：
```python
# 列表推导式 — 一次性生成所有数据
squares = [x**2 for x in range(1000000)] # 占用大量内存
# 生成器表达式 — 按需生成
squares = (x**2 for x in range(1000000)) # 几乎不占内存
# 在 sum/max/min 等函数里直接用
total = sum(x**2 for x in range(1000000))
```
处理大数据时，生成器表达式是列表推导式的直接替代。
## yield from：委托给子生成器
`yield from` 把迭代委托给另一个生成器，避免手写 for 循环：
```python
def flatten(nested):
for item in nested:
if isinstance(item, list):
yield from flatten(item) # 递归展开
else:
yield item
list(flatten([1, [2, [3, 4]], 5])) # [1, 2, 3, 4, 5]
```
`yield from` 不只是语法糖——它还正确处理了 `send()`、`throw()`、`close()` 等生成器方法的传递。
## 生成器做协程
`yield` 不只能返回值，还能接收值（通过 `send()`），这让生成器可以用来实现协程：
```python
def accumulator():
total = 0
while True:
value = yield total
if value is None:
break
total += value
gen = accumulator()
next(gen) # 启动生成器，返回 0
gen.send(10) # 返回 10
gen.send(20) # 返回 30
```
Python 3.5+ 推荐用 `async/await` 替代这种用法，但理解 `yield` 的双向通信有助于理解协程原理。
## 追问
### 迭代器只能遍历一次吗？
是的。迭代器是有状态的，遍历完就空了。要重新遍历，需要创建新的迭代器。可迭代对象（如列表）每次调用 `iter()` 都返回新的迭代器。
### 生成器的内存优势有多大？
读 10GB 日志文件，`for line in open("log.txt")` 只占几 KB 内存（每次读一行）。`readlines()` 会把整个文件加载到内存。差距在数据量大时非常显著。服务端6月2日 01:36
Python 内存管理是怎样的？引用计数、分代 GC 和内存池原理Python 内存管理分三层：引用计数（主要）、垃圾回收（处理循环引用）、内存池（减少 malloc 开销）。日常开发不需要手动管理内存，但理解机制能帮你排查内存泄漏。
## 引用计数：核心机制
每个对象维护一个引用计数 `ob_refcnt`。引用增加时 +1，引用减少时 -1，归零时立即释放。
```python
import sys
a = [1, 2, 3] # 引用计数 1
b = a # 引用计数 2
c = a # 引用计数 3
print(sys.getrefcount(a)) # 4（多 1 是因为 getrefcount 参数本身也是引用）
del b # 引用计数 2
c = None # 引用计数 1
# a 离开作用域后引用计数归零，内存释放
```
引用计数的优势：实时释放，不需要暂停程序做垃圾回收。劣势：无法处理循环引用。
## 循环引用问题
```python
a = []
b = []
a.append(b) # a 引用 b
b.append(a) # b 引用 a
del a, b # 引用计数各剩 1（互相引用），永远不会归零
```
引用计数对循环引用无能为力。Python 用分代垃圾回收（GC）处理这种情况。
## 分代垃圾回收
GC 把对象分成三代：第 0 代（新对象）、第 1 代、第 2 代（长寿对象）。
- 新创建的对象在第 0 代
- 经过一次 GC 存活的对象晋升到下一代
- 第 0 代 GC 最频繁（阈值约 700 个对象），第 2 代最少
分代回收的理论依据：大部分对象很快变成垃圾（如函数内的临时变量），长寿对象倾向于一直活着。只频繁检查年轻对象，减少 GC 开销。
```python
import gc
print(gc.get_threshold()) # (700, 10, 10) — 第0代阈值700，每10次第0代GC触发1次第1代
print(gc.get_count()) # 当前各代对象计数
```
手动触发 GC：`gc.collect()`。通常不需要手动调用，但在处理大量循环引用对象后可以主动回收。
## 内存池：pymalloc
Python 不直接用系统的 malloc/free 管理小对象，而是用自己实现的 pymalloc 内存池：
- **小对象（<512 字节）**：由 pymalloc 从大块内存中分配，减少系统调用
- **大对象（>=512 字节）**：直接用系统 malloc
内存池按 256 KB 的 arena 分块，arena 内按 4 KB 的 pool 分块，pool 内按固定大小的 block 分配。相同大小的 block 共享 pool，减少碎片。
这就是为什么 Python 进程的 RSS（常驻内存）不会随着对象释放而下降——pymalloc 保留空闲 arena 供后续使用，不归还给操作系统。
## 内存泄漏排查
Python 的内存泄漏通常是"对象被意外引用导致无法释放"而非真正的泄漏。
**1. 用 objgraph 找到引用链**
```python
import objgraph
objgraph.show_backrefs(objgraph.by_type("dict")[0], max_depth=5)
```
生成引用关系图，找到谁在持有不该持有的引用。
**2. 用 tracemalloc 定位分配位置**
```python
import tracemalloc
tracemalloc.start()
# ... 运行代码 ...
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")
for stat in top_stats[:10]:
print(stat)
```
显示内存分配最多的代码行。
**3. 常见泄漏原因**
- 全局列表/字典不断追加但从不清理
- 闭包或回调持有大对象引用
- `__del__` 方法导致对象无法被 GC 回收（Python 3.4 已改善）
- 缓存没有上限（用 `functools.lru_cache` 替代手动缓存）
## 追问
### 为什么 Python 进程的内存只增不减？
pymalloc 保留空闲 arena 不归还操作系统。如果你看到 RSS 很高但对象不多，可能是内存碎片。`malloc_trim(0)` 可以强制归还（Linux），但不保证生效。
### 引用计数和 GC 哪个先执行？
引用计数是实时的，每次赋值/删除都更新。GC 是周期性运行的，只处理循环引用。两者配合工作。服务端6月2日 01:35
Python 异常处理怎么写？try/except/else/finally 和自定义异常详解Python 用 try/except 捕获异常，else 放无异常时执行的代码，finally 放无论如何都执行的清理逻辑。自定义异常继承 Exception，让错误类型可区分。
## 基本结构
```python
try:
result = 10 / 0
except ZeroDivisionError:
print("除零错误")
except (TypeError, ValueError) as e:
print(f"类型或值错误: {e}")
else:
print("没有异常时执行")
finally:
print("无论如何都执行")
```
- `else` 在 try 块没有抛异常时执行，适合放"成功后才做的事"
- `finally` 总是执行，即使 try 里有 return 或 except 里有 raise。用于释放资源（关闭文件、断开连接）
## 异常继承体系
所有异常继承自 `BaseException`，日常使用继承 `Exception`：
```
BaseException
+-- SystemExit # sys.exit()
+-- KeyboardInterrupt # Ctrl+C
+-- Exception # 所有业务异常的基类
+-- ValueError
+-- TypeError
+-- KeyError
+-- FileNotFoundError
```
不要捕获 `BaseException`——`except:` 或 `except BaseException:` 会把 `KeyboardInterrupt` 和 `SystemExit` 也吞掉，程序无法正常退出。永远用 `except Exception:` 或更具体的异常类型。
## 自定义异常
```python
class BusinessError(Exception):
"""业务逻辑错误基类"""
pass
class InsufficientBalance(BusinessError):
def __init__(self, balance, amount):
self.balance = balance
self.amount = amount
super().__init__(f"余额不足: 余额 {balance}, 需要 {amount}")
try:
withdraw(account, amount)
except InsufficientBalance as e:
print(e)
print(e.balance) # 100
```
自定义异常比直接 `raise ValueError("余额不足")` 好在：调用方可以按类型捕获（`except InsufficientBalance`），而不是解析错误消息字符串。
## 异常链：raise from
捕获一个异常后抛出另一个，保留原始异常的堆栈：
```python
try:
data = json.loads(raw)
except json.JSONDecodeError as e:
raise DataFormatError("数据格式错误") from e
```
`from e` 让新异常的 `__cause__` 指向原始异常。调试时能看到完整的异常链。不加 `from e`，原始异常的堆栈信息会丢失。
## 常见反模式
**1. 空 except 吞掉所有错误**
```python
try:
do_something()
except: # 错误！吞掉 KeyboardInterrupt
pass
# 正确做法
except Exception as e:
log.error(f"操作失败: {e}")
```
**2. except 块太宽**
```python
try:
value = data["key"]
result = int(value)
except Exception: # 太宽，KeyError 和 ValueError 混在一起
...
# 正确：分别处理
except KeyError:
result = 0
except ValueError:
result = -1
```
**3. 用异常做流程控制**
```python
# 错误：用异常判断 key 是否存在
try:
value = d[key]
except KeyError:
value = default
# 正确：用 get 或 in
value = d.get(key, default)
```
异常比条件判断慢 10-100 倍。只对真正的"异常"情况用异常，正常流程用条件判断。
## 追问
### finally 里的 return 会覆盖 try 里的 return 吗？
会。永远不要在 finally 里写 return。
```python
def foo():
try:
return 1
finally:
return 2 # 覆盖 try 里的
print(foo()) # 2
```服务端6月2日 01:33
Python 装饰器高级用法有哪些？带参数装饰器、类装饰器和 functools.wraps 详解装饰器的高级用法围绕三个问题：怎么传参数、怎么保持被装饰函数的元信息、什么时候用类而不是函数写装饰器。
## 带参数的装饰器
普通装饰器只能装饰函数，不能接收额外参数。需要参数时，加一层嵌套：
```python
def retry(max_attempts=3, delay=1):
def decorator(func):
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
time.sleep(delay)
return wrapper
return decorator
@retry(max_attempts=5, delay=2)
def call_api():
...
```
`@retry(5, 2)` 先调用 `retry(5, 2)` 返回 `decorator`，再 `decorator(call_api)` 返回 `wrapper`。三层嵌套是带参数装饰器的固定模式。
## functools.wraps：保持函数身份
装饰器替换了原函数，导致 `__name__`、`__doc__` 等元信息丢失：
```python
def log(func):
def wrapper(*args, **kwargs):
print(f"Calling {func.__name__}")
return func(*args, **kwargs)
return wrapper
@log
def hello(): pass
print(hello.__name__) # 'wrapper' 而不是 'hello'
```
加 `@wraps(func)` 把原函数的元信息复制到 wrapper：
```python
from functools import wraps
def log(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"Calling {func.__name__}")
return func(*args, **kwargs)
return wrapper
print(hello.__name__) # 'hello'
```
`@wraps` 必须加——调试时看到 `wrapper` 而不是实际函数名，排查问题会非常痛苦。
## 类装饰器
用类写装饰器，适合需要维护状态的场景：
```python
class CountCalls:
def __init__(self, func):
self.func = func
self.count = 0
wraps(func)(self) # 保持元信息
def __call__(self, *args, **kwargs):
self.count += 1
print(f"{self.func.__name__} called {self.count} times")
return self.func(*args, **kwargs)
@CountCalls
def say_hello():
print("Hello")
say_hello() # say_hello called 1 times; Hello
say_hello() # say_hello called 2 times; Hello
```
`__init__` 接收被装饰的函数，`__call__` 每次调用时执行。`self.count` 跨调用持久化——函数写装饰器用闭包变量存状态，类装饰器用实例属性，语义更清晰。
## 装饰类
装饰器不只能装饰函数，还能装饰整个类：
```python
def add_repr(cls):
def __repr__(self):
attrs = ', '.join(f'{k}={v!r}' for k, v in self.__dict__.items())
return f'{cls.__name__}({attrs})'
cls.__repr__ = __repr__
return cls
@add_repr
class User:
def __init__(self, name, age):
self.name = name
self.age = age
print(User('Alice', 30)) # User(name='Alice', age=30)
```
dataclass 的 `@dataclass` 就是类装饰器——自动生成 `__init__`、`__repr__`、`__eq__`。
## 常见的高级装饰器模式
**1. 缓存/记忆化**
```python
from functools import lru_cache
@lru_cache(maxsize=128)
def expensive(n):
return sum(i ** 2 for i in range(n))
```
标准库自带，不用自己写。`maxsize=None` 无限缓存。
**2. 类型检查**
```python
def validate(**types):
def decorator(func):
@wraps(func)
def wrapper(**kwargs):
for name, expected in types.items():
if name in kwargs and not isinstance(kwargs[name], expected):
raise TypeError(f'{name} must be {expected}')
return func(**kwargs)
return wrapper
return decorator
@validate(age=int, name=str)
def create_user(name, age):
...
```
**3. 超时控制**
```python
import signal
def timeout(seconds):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
def handler(signum, frame):
raise TimeoutError(f'{func.__name__} timed out')
old = signal.signal(signal.SIGALRM, handler)
signal.alarm(seconds)
try:
return func(*args, **kwargs)
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old)
return wrapper
return decorator
```
## 追问
### 多个装饰器的执行顺序？
从下到上装饰，从上到下执行。`@a @b def f()` 等价于 `a(b(f))`，调用 `f()` 时先执行 `a` 的逻辑，再执行 `b` 的逻辑，最后执行 `f`。
### 装饰器和 AOP 是什么关系？
装饰器是 Python 实现 AOP（面向切面编程）的方式。日志、权限、缓存这些"横切关注点"用装饰器统一处理，不侵入业务代码。服务端6月2日 01:32
Python 元类是什么？type 怎么创建类？元类什么时候该用？元类是创建类的类。普通类创建实例，元类创建类。Python 里 `class` 语句本质上是调用 `type()` 来创建类对象，元类让你拦截这个过程，在类创建时自动修改类的属性、方法、继承关系。
## Python 类的创建过程
```python
class Foo:
x = 1
```
这行代码执行时，Python 做了这件事：
```python
Foo = type('Foo', (object,), {'x': 1})
```
`type(类名, 父类元组, 属性字典)` 就是创建类的底层调用。`type` 本身就是一个元类——所有类都是 `type` 的实例。
```python
print(type(Foo)) # <class 'type'>
print(type(Foo())) # <class '__main__.Foo'>
```
`type(Foo)` 是 `type`，说明 Foo 类是 type 的实例。`type(Foo())` 是 Foo，说明 Foo() 实例是 Foo 的实例。
## 自定义元类
继承 `type`，重写 `__new__` 或 `__init__`：
```python
class Meta(type):
def __new__(cls, name, bases, dct):
# 在类创建前修改属性字典
dct['created_by_meta'] = True
return super().__new__(cls, name, bases, dct)
class Foo(metaclass=Meta):
x = 1
print(Foo.created_by_meta) # True
```
`__new__` 在类对象创建之前调用，可以修改 `dct`（属性字典）。`__init__` 在类对象创建之后调用，可以修改已创建的类。大多数场景用 `__new__` 就够了。
## 实际用途
**1. 自动注册子类**
ORM、插件系统常用——每定义一个子类就自动注册到一个全局字典里：
```python
class RegistryMeta(type):
_registry = {}
def __new__(cls, name, bases, dct):
klass = super().__new__(cls, name, bases, dct)
if name != 'Base': # 跳过基类本身
cls._registry[name] = klass
return klass
class Base(metaclass=RegistryMeta):
pass
class User(Base): pass
class Order(Base): pass
print(RegistryMeta._registry) # {'User': <class 'User'>, 'Order': <class 'Order'>}
```
Django 的 ModelBase 就是这种模式——每个 Model 子类自动注册到 Django 的 app registry。
**2. 接口检查**
强制子类实现特定方法：
```python
class InterfaceMeta(type):
def __new__(cls, name, bases, dct):
if bases: # 不是基类本身
required = getattr(bases[0], '_required_methods', [])
for method in required:
if method not in dct:
raise TypeError(f'{name} 必须实现 {method}')
return super().__new__(cls, name, bases, dct)
class Animal(metaclass=InterfaceMeta):
_required_methods = ['speak']
class Dog(Animal):
def speak(self): return 'Woof!'
class Cat(Animal): pass # TypeError: Cat 必须实现 speak
```
**3. 单例模式**
```python
class SingletonMeta(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class Database(metaclass=SingletonMeta):
pass
```
重写 `__call__` 拦截实例化过程——第一次创建实例，后续返回同一个。
## 追问
### 元类和类装饰器有什么区别？
类装饰器在类创建之后修改，元类在类创建过程中修改。类装饰器更简单透明，优先用装饰器。元类能控制 `__new__` 和 `__init__`，能做的事更多但也更难调试。
### 什么时候该用元类？
几乎不需要。Python 社区有一句话："如果你不确定是否需要元类，你就不需要。"95% 的场景用类装饰器、`__init_subclass__`、或普通继承就能解决。元类适合框架作者（Django ORM、SQLAlchemy），应用层开发者很少需要。服务端6月2日 01:31
Python 闭包是什么？变量怎么被记住的？闭包是一个函数记住了自己被创建时的作用域里的变量，即使那个作用域已经执行完毕。Python 里闭包最常见的用途：工厂函数、延迟计算、替代简单类。
## 闭包怎么产生的
```python
def make_counter(start=0):
count = start
def counter():
nonlocal count
count += 1
return count
return counter
c = make_counter(10)
print(c()) # 11
print(c()) # 12
```
`make_counter` 执行完了，`count` 变量按理应该被销毁。但 `counter` 函数内部引用了 `count`，Python 会把 `count` 和 `counter` 绑在一起——这就是闭包。`counter` 闭包了 `count` 变量。
`nonlocal` 声明告诉 Python `count` 不是局部变量，而是外层作用域的变量。不加 `nonlocal`，`count += 1` 会在 `counter` 内部创建一个新的局部变量 `count`，而不是修改外层的。
## 闭包存储在哪里
闭包变量存在函数的 `__closure__` 属性里：
```python
print(c.__closure__) # (<cell at 0x...: int object at ...>,)
print(c.__closure__[0].cell_contents) # 12
```
每个被闭包的变量是一个 cell 对象。这就是 Python 实现闭包的底层机制。
## 常见用途
**1. 工厂函数**
根据参数生成不同的函数：
```python
def power(exp):
def f(base):
return base ** exp
return f
square = power(2)
cube = power(3)
print(square(5)) # 25
print(cube(5)) # 125
```
**2. 缓存/记忆化**
```python
def memoize(fn):
cache = {}
def wrapper(*args):
if args not in cache:
cache[args] = fn(*args)
return cache[args]
return wrapper
@memoize
def fib(n):
if n < 2: return n
return fib(n-1) + fib(n-2)
print(fib(100)) # 瞬间出结果
```
`cache` 被 `wrapper` 闭包，不需要全局变量。这就是装饰器能工作的基础——装饰器本质上就是闭包。
**3. 替代简单类**
只有状态 + 一个方法的场景，闭包比类轻量：
```python
# 闭包方式
def make_accumulator(initial=0):
total = initial
def add(value):
nonlocal total
total += value
return total
return add
acc = make_accumulator()
print(acc(10)) # 10
print(acc(20)) # 30
```
## 追问
### 闭包和类怎么选？
有多个方法或复杂状态管理用类。只有一个操作 + 简单状态用闭包。闭包更函数式，类更面向对象。
### 闭包会导致内存泄漏吗？
会。被闭包的变量不会在函数返回后释放，只要闭包函数还活着，变量就一直在。长期存活的闭包（如事件监听器、回调）要注意。
### lambda 能形成闭包吗？
能。`lambda x: x + offset` 里的 `offset` 就是被闭包的变量。但 lambda 不能用 `nonlocal`，所以无法修改外层变量，只能读取。服务端5月29日 00:09
Python 多线程和多进程有什么区别？GIL 对多线程有什么影响？核心区别：进程是资源分配单位，有独立内存空间；线程是调度单位，共享进程内存。在 Python 里这道题的特殊之处是 GIL——全局解释器锁让同一时刻只有一个线程执行 Python 字节码，所以 CPU 密集型任务用多线程不会加速，必须用多进程。I/O 密集型任务多线程够用，因为等 I/O 时 GIL 会释放。
## 追问
### GIL 到底锁的是什么？为什么不能去掉？
GIL 保护的是 CPython 的内存管理——引用计数。CPython 的对象引用计数不是线程安全的，如果多个线程同时修改引用计数，对象可能被提前释放或泄漏。加锁是最简单的方案，代价是多线程无法真正并行。Python 3.13 开始实验性支持 free-threaded 模式（PEP 703），试图移除 GIL，但目前生态兼容性还是问题。短期内 GIL 不会消失。
### 多线程既然不能并行，还有用吗？
有用。线程等待 I/O（网络请求、文件读写、数据库查询）时会释放 GIL，其他线程可以执行。所以 I/O 密集型场景（爬虫、API 调用、数据库操作）用多线程完全没问题。实测：100 个 HTTP 请求，单线程串行 10 秒，10 线程并发约 1.2 秒，接近 10 倍加速。但如果是纯计算（比如大矩阵运算），10 线程和 1 线程耗时几乎一样，甚至更慢——线程切换本身有开销。
### 什么时候用多进程？有什么坑？
CPU 密集型任务：数据处理、图像处理、机器学习训练。`multiprocessing.Pool` 是最常用的接口。最大的坑是进程间通信开销——进程不共享内存，数据要序列化传输。传一个大 numpy 数组给子进程，pickle 序列化的时间可能比计算本身还长。解决方案：用 `multiprocessing.shared_memory`（Python 3.8+）或 `multiprocessing.Array` 共享内存，避免序列化。另一个坑是子进程的异常不会自动传播到主进程，需要手动处理。
### 协程和多线程有什么区别？
协程是用户态的协作式调度，线程是内核态的抢占式调度。协程的切换由代码控制（await），线程的切换由操作系统控制。协程没有锁的问题——同一个时刻只有一个协程在执行，不存在竞态条件。代价是协程里不能有阻塞调用，一阻塞整个事件循环就卡住了。asyncio 适合高并发 I/O（上万连接的聊天服务器），threading 适合少量 I/O 并发或需要兼容阻塞库的场景。
## 写段代码
```python
from multiprocessing import Pool
import time
def heavy_compute(n):
return sum(i * i for i in range(n))
if __name__ == '__main__':
# 多进程：4核并行，接近4倍加速
with Pool(4) as p:
results = p.map(heavy_compute, [10**7] * 4)
print(results)
# 对比：多线程对CPU密集型无效
# 4个线程和1个线程耗时几乎一样（GIL）
```服务端5月29日 00:07
Python 描述符是什么？数据描述符和非数据描述符优先级怎么排？描述符是实现了 `__get__`、`__set__`、`__delete__` 中任意一个的类，被赋值给另一个类的类属性后，会拦截那个属性的访问。Python 的属性查找有一套隐藏规则：当解释器在类（及其 MRO）的 `__dict__` 里找到的值是描述符时，不会直接返回它，而是调用描述符的 `__get__` 方法。这就是 `property`、`classmethod`、`staticmethod` 的底层原理——它们都是描述符。
## 追问
### 数据描述符和非数据描述符有什么区别？优先级怎么排？
关键区别是有没有 `__set__`。实现了 `__get__` + `__set__` 的叫数据描述符，只有 `__get__` 的叫非数据描述符。优先级：数据描述符 > 实例 `__dict__` > 非数据描述符。换句话说，数据描述符能拦截赋值操作，实例 `__dict__` 里写不进去；非数据描述符拦截不了，一旦实例 `__dict__` 有了同名 key 就被覆盖了。这就是为什么 `property`（数据描述符）设了 setter 后 `obj.x = 1` 一定走 setter，而普通方法（非数据描述符）可以被实例属性遮蔽。
### Python 属性查找的完整顺序是什么？
按这个顺序：1. 类及其 MRO 的 `__dict__` 里找，如果是数据描述符就调 `__get__` 返回；2. 实例 `__dict__` 里找；3. 回到类的 `__dict__`，如果是非数据描述符就调 `__get__` 返回。这解释了一个经典面试题：为什么实例能覆盖普通方法但覆盖不了 property？因为方法是非数据描述符，步骤 2 的实例 `__dict__` 优先级更高；property 是数据描述符，步骤 1 就截走了。
### __set_name__ 是什么？为什么需要它？
Python 3.6 新增的钩子。描述符被赋值到类属性时，解释器自动调用 `desc.__set_name__(owner, name)`，把属性名传进去。之前描述符不知道自己叫什么名字，要么手动传（`age = Typed('age', int)`），要么用元类扫描类 `__dict__` 来推断。有了 `__set_name__`，Django ORM 的 `name = CharField()` 就不用重复写字段名了——`CharField.__set_name__` 会自动收到 `'name'`。
### 描述符里怎么存值？为什么不能直接用 self.xxx？
描述符实例是类级别的，所有实例共享同一个描述符对象。如果你在 `__set__` 里写 `self.value = val`，所有实例共享同一个 value，后面的赋值会覆盖前面的。正确做法是存到 `obj.__dict__[self.name]` 里，或者用 `weakref.WeakKeyDictionary` 做 `self` → `value` 的映射。前一种更常见（property 就这么做的），后一种适合描述符本身需要维护额外状态的场景。
## 写段代码
```python
# 用 __set_name__ 实现类型检查描述符
class Typed:
def __init__(self, expected_type):
self.expected_type = expected_type
def __set_name__(self, owner, name):
self.name = name # 自动获取属性名
self.storage = f'_{name}'
def __get__(self, obj, objtype=None):
if obj is None:
return self
return getattr(obj, self.storage, None)
def __set__(self, obj, value):
if not isinstance(value, self.expected_type):
raise TypeError(f'{self.name} 需要 {self.expected_type.__name__}')
setattr(obj, self.storage, value)
class User:
name = Typed(str)
age = Typed(int)
u = User()
u.name = "Alice" # OK
u.age = 25 # OK
# u.age = "25" # TypeError: age 需要 int
```服务端5月29日 00:06
Python 列表推导式和生成器表达式有什么区别？什么时候该用哪个？区别就一个字：方括号 `[]` 立即算出所有结果放进列表，圆括号 `()` 返回一个生成器对象，用到哪个才算哪个。`[x**2 for x in range(10)]` 执行完内存里就有 10 个数；`(x**2 for x in range(10))` 执行完只多了一个 200 字节的生成器对象，值还没算。
## 追问
### 生成器只能遍历一次，踩过坑吗？
这是最常见的陷阱。你写 `gen = (x for x in range(5))`，第一次 `list(gen)` 得到 `[0,1,2,3,4]`，再 `list(gen)` 就是 `[]`——生成器耗尽了。如果后面的代码还要用，要么转成列表存起来，要么重新创建生成器。调试时这个坑尤其烦人：你在调试器里 `print(list(gen))` 看了一眼，后面代码就拿不到数据了。
### sum(x**2 for x in range(N)) 和 sum([x**2 for x in range(N)]) 哪个快？
大多数人觉得生成器快，实际不一定。生成器省内存是肯定的，但每次 yield 有函数调用开销。列表推导式的循环在 C 层执行（CPython 实现中 listcomp 是专用字节码），而生成器每次 yield 要切换栈帧。数据量小时列表版反而更快——省下的内存分配开销比 yield 开销小。数据量大时生成器版才赢，因为列表版要先把所有结果存内存。经验值：N < 1000 差别可以忽略，N > 10 万生成器才有明显优势。
### 字典推导式和集合推导式呢？
语法一样，换括号就行：`{k: v for k, v in pairs}` 是字典推导式，`{x for x in items}` 是集合推导式。注意集合推导式没有生成器版本——`{x for x in items}` 是立即求值的，没有惰性集合。如果需要惰性去重，得用生成器 + `set()` 分两步。
### 嵌套推导式怎么读？
从左到右读，和嵌套 for 循环的顺序一致。`[f(x,y) for x in A for y in B]` 等价于 `for x in A: for y in B: f(x,y)`。超过两层就该换普通循环了，没人能在脑内解析三层推导式。Python 之禅说得明白：可读性很重要。
## 写段代码
```python
# 生成器只能遍历一次的坑
gen = (x**2 for x in range(5))
print(list(gen)) # [0, 1, 4, 9, 16]
print(list(gen)) # [] ← 耗尽了！
# 需要多次使用就转列表
squares = [x**2 for x in range(5)]
print(squares[:3]) # [0, 1, 4]
print(squares[3:]) # [9, 16]
# 管道式处理用生成器省内存
lines = (line.strip() for line in open('big.log')) # 不读全文
errors = (line for line in lines if 'ERROR' in line)
count = sum(1 for _ in errors) # 只计数，不存结果
```服务端5月29日 00:05
Python 面向对象的核心概念有哪些？MRO 和描述符怎么理解？Python 面向对象的核心是四件事：用类组织数据和行为的封装机制、通过继承复用代码、用多态让不同对象响应同一接口、以及 Python 自己的特殊之处——MRO、描述符、__slots__ 这些面试高频考点。基础概念（类/实例/属性/方法）不展开，下面只说容易踩坑和被追问的部分。
## 追问
### Python 的 MRO 是怎么排的？为什么不用深度优先？
Python 3 用 C3 线性化算法计算 MRO。核心规则：子类排在父类前面，同一层按定义顺序排，不能违反前两条规定。为什么不用深度优先？因为菱形继承下深度优先会重复访问基类。经典例子：D 继承 B 和 C，B 和 C 都继承 A，深度优先的顺序是 D→B→A→C→A，A 被访问两次。C3 的结果是 D→B→C→A，每个类只出现一次，且 B 在 C 前面（定义顺序）。通过 `ClassName.__mro__` 可以查看任意类的解析顺序。
### __slots__ 能省多少内存？有什么代价？
普通 Python 对象用 `__dict__` 存属性，一个空对象就要占 56 字节（64 位 CPython）。`__slots__` 用固定数组替代字典，属性直接按偏移量访问，省掉哈希表开销。实际测量：100 万个只有 name 和 age 属性的对象，用 `__dict__` 约 160MB，用 `__slots__` 约 48MB，省 70%。代价是不能再动态添加属性，而且继承时如果父类没有声明 `__slots__`，子类照样会有 `__dict__`，优化白做。实际项目中，Django 的 QuerySet 用了 `__slots__` 优化大量小对象。
### 描述符是什么？property 和 classmethod 跟它什么关系？
描述符是实现了 `__get__`、`__set__`、`__delete__` 中任意一个的类。Python 的属性查找有个隐藏步骤：如果找到的对象是描述符，就调用它的 `__get__` 返回结果，而不是直接返回对象本身。`property` 就是描述符——你的 getter/setter 被 `__get__`/`__set__` 包装了；`classmethod` 也是描述符——它的 `__get__` 把类传给函数而不是实例。区分数据描述符（有 `__set__`）和非数据描述符（只有 `__get__`）：数据描述符优先级高于实例 `__dict__`，非数据描述符优先级低于实例 `__dict__`。这就是为什么 property 能拦截赋值而普通方法不行。
### __new__ 和 __init__ 有什么区别？
`__new__` 创建对象并返回，`__init__` 初始化已创建的对象。`__new__` 是类方法（第一个参数是 cls），`__init__` 是实例方法（第一个参数是 self）。单例模式用 `__new__` 控制：如果 `_instance` 已存在就直接返回，不再创建新对象。`__init__` 做不到这点——它执行时对象已经创建了。另一个场景：不可变类型（str、int、tuple）的子类化必须重写 `__new__`，因为这些类型的对象在 `__new__` 阶段就已经确定了值，`__init__` 改不了。
## 写段代码
```python
# 描述符实现懒加载属性
class LazyProperty:
def __init__(self, func):
self.func = func
def __get__(self, obj, cls):
if obj is None:
return self
value = self.func(obj)
obj.__dict__[self.func.__name__] = value # 缓存到实例字典
return value
class Data:
@LazyProperty
def expensive(self):
print("计算中...")
return sum(range(1000000))
d = Data()
print(d.expensive) # 计算中... 499999500000
print(d.expensive) # 499999500000（不再计算，从 __dict__ 直接取）
```