1
import functools
2

3
# 装饰器本质：接收函数并返回增强函数的函数
4
def timer(func):
5
    @functools.wraps(func)  # 保留原函数元数据
6
    def wrapper(*args, **kwargs):
7
        import time
8
        start = time.perf_counter()
9
        result = func(*args, **kwargs)
10
        end = time.perf_counter()
11
        print(f"{func.__name__} 耗时: {end - start:.4f}s")
12
        return result
13
    return wrapper
14

15
@timer
16
def slow_function():
17
    import time
18
    time.sleep(1)
19

20
# 等价于：slow_function = timer(slow_function)

1
import functools
2
import time
3

4
# 带参数的装饰器：返回装饰器的函数
5
def retry(max_attempts=3, delay=1):
6
    def decorator(func):
7
        @functools.wraps(func)
8
        def wrapper(*args, **kwargs):
9
            last_exception = None
10
            for attempt in range(max_attempts):
11
                try:
12
                    return func(*args, **kwargs)
13
                except Exception as e:
14
                    last_exception = e
15
                    if attempt < max_attempts - 1:
16
                        time.sleep(delay)
17
            raise last_exception
18
        return wrapper
19
    return decorator
20

21
@retry(max_attempts=3, delay=2)
22
def unstable_api():
23
    pass

1
import functools
2

3
# 类装饰器：实现 __call__ 方法
4
class CallCounter:
5
    def __init__(self, func):
6
        self.func = func
7
        self.count = 0
8
        functools.update_wrapper(self, func)
9

10
    def __call__(self, *args, **kwargs):
11
        self.count += 1
12
        print(f"调用次数: {self.count}")
13
        return self.func(*args, **kwargs)
14

15
@CallCounter
16
def my_function():
17
    pass
18

19
# 使用场景：缓存、计时、访问控制
20
class Memoize:
21
    def __init__(self, func):
22
        self.func = func
23
        self.cache = {}
24
        functools.update_wrapper(self, func)
25

26
    def __call__(self, *args):
27
        if args not in self.cache:
28
            self.cache[args] = self.func(*args)
29
        return self.cache[args]

1
# 装饰器从下到上执行
2
@decorator1  # 第2个执行
3
@decorator2  # 第1个执行
4
def func():
5
    pass
6

7
# 等价于：func = decorator1(decorator2(func))

1
# 生成器函数：包含 yield 关键字
2
def fibonacci(n):
3
    a, b = 0, 1
4
    for _ in range(n):
5
        yield a
6
        a, b = b, a + b
7

8
# 生成器对象
9
gen = fibonacci(10)
10
print(type(gen))  # <class 'generator'>
11

12
# 迭代获取值
13
for value in gen:
14
    print(value)
15

16
# next() 获取
17
gen = fibonacci(5)
18
print(next(gen))  # 0
19
print(next(gen))  # 1
20
print(next(gen))  # 1

1
# 生成器表达式：惰性求值
2
gen = (x**2 for x in range(1000000))  # 不占满内存
3

4
# 列表推导式：立即求值
5
lst = [x**2 for x in range(1000000)]  # 占用大量内存
6

7
# 使用场景对比
8
# 生成器适合：大数据处理、管道处理
9
# 列表适合：小数据、需要多次迭代
10

11
# 链接生成器
12
def numbers():
13
    for i in range(10):
14
        yield i
15

16
def evens():
17
    for i in numbers():
18
        if i % 2 == 0:
19
            yield i
20

21
# itertools 生成器
22
from itertools import count, cycle, chain, islice, takewhile
23

24
# 无限计数器
25
for i in islice(count(10), 5):  # 10, 11, 12, 13, 14
26
    print(i)
27

28
# 循环
29
for i in islice(cycle(['A', 'B']), 5):  # A, B, A, B, A
30
    print(i)
31

32
# 链式迭代
33
for i in chain([1, 2], ['a', 'b']):  # 1, 2, a, b
34
    print(i)

1
# 迭代器必须实现 __iter__ 和 __next__
2
class Counter:
3
    def __init__(self, n):
4
        self.n = n
5
        self.current = 0
6

7
    def __iter__(self):
8
        return self
9

10
    def __next__(self):
11
        if self.current >= self.n:
12
            raise StopIteration
13
        result = self.current
14
        self.current += 1
15
        return result
16

17
for i in Counter(3):
18
    print(i)  # 0, 1, 2

1
# send() 与生成器通信
2
def coroutine():
3
    result = None
4
    while True:
5
        received = yield result
6
        result = received * 2
7

8
co = coroutine()
9
next(co)  # 启动协程
10
print(co.send(10))  # 20
11
print(co.send(5))   # 10
12

13
# close() 关闭协程
14
co.close()
15

16
# throw() 抛出异常
17
try:
18
    co.throw(ValueError, ValueError("error"))
19
except ValueError:
20
    pass

1
# 实现 __enter__ 和 __exit__
2
class DatabaseConnection:
3
    def __enter__(self):
4
        print("连接数据库")
5
        return self
6

7
    def __exit__(self, exc_type, exc_val, exc_tb):
8
        print("关闭数据库连接")
9
        # 返回 True 抑制异常
10
        return False
11

12
with DatabaseConnection() as db:
13
    print("使用数据库")
14

15
# 异常处理
16
class SafeConnection:
17
    def __enter__(self):
18
        return self
19

20
    def __exit__(self, exc_type, exc_val, exc_tb):
21
        if exc_type is not None:
22
            print(f"异常: {exc_val}")
23
            return False  # 不抑制异常
24
        return True

1
from contextlib import contextmanager
2

3
# 使用生成器实现上下文管理器
4
@contextmanager
5
def transaction(connection):
6
    print("开始事务")
7
    try:
8
        yield "tx_handle"
9
        print("提交事务")
10
    except Exception:
11
        print("回滚事务")
12
        raise
13

14
with transaction(conn) as tx:
15
    pass
16

17
# 清理操作
18
@contextmanager
19
def managed_resource(name):
20
    print(f"获取 {name}")
21
    try:
22
        yield name
23
    finally:
24
        print(f"释放 {name}")

1
from contextlib import ExitStack
2

3
# 多个上下文管理器
4
with ExitStack() as stack:
5
    conn1 = stack.enter_context(db1_connection())
6
    conn2 = stack.enter_context(db2_connection())
7
    # 自动管理所有资源
8

9
# 条件进入
10
with ExitStack() as stack:
11
    if needs_cache:
12
        cache = stack.enter_context(CacheManager())
13
    if needs_logging:
14
        logger = stack.enter_context(Logger())

1
from functools import partial
2

3
# 固定函数的部分参数
4
def power(base, exponent):
5
    return base ** exponent
6

7
square = partial(power, exponent=2)
8
cube = partial(power, exponent=3)
9

10
print(square(5))  # 25
11
print(cube(5))     # 125
12

13
# 常用场景：回调函数
14
from functools import partial
15

16
def send_email(to, subject, body):
17
    print(f"To: {to}, Subject: {subject}")
18

19
send_to_alice = partial(send_email, to="alice@example.com")
20
send_to_alice(subject="Hello", body="Hi!")

1
# __call__ 使对象可像函数一样调用
2
class Adder:
3
    def __init__(self, n):
4
        self.n = n
5

6
    def __call__(self, x):
7
        return self.n + x
8

9
add5 = Adder(5)
10
print(add5(10))  # 15
11
print(callable(add5))  # True
12

13
# 使用场景：带状态的函数
14
class RateLimiter:
15
    def __init__(self, max_calls, period):
16
        self.max_calls = max_calls
17
        self.period = period
18
        self.calls = []
19

20
    def __call__(self, func):
21
        import time
22
        now = time.time()
23
        self.calls = [t for t in self.calls if now - t < self.period]
24
        if len(self.calls) >= self.max_calls:
25
            raise Exception("Rate limit exceeded")
26
        self.calls.append(now)
27
        return func()

1
from functools import lru_cache
2

3
@lru_cache(maxsize=128)
4
def expensive_computation(n):
5
    import time
6
    time.sleep(1)
7
    return n * n
8

9
# 缓存管理
10
print(expensive_computation.cache_info())
11
# CacheInfo(hits=0, misses=0, maxsize=128, currsize=0)
12

13
expensive_computation(10)  # 计算
14
expensive_computation(10)  # 使用缓存
15
print(expensive_computation.cache_info())
16
# CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)
17

18
# 清除缓存
19
expensive_computation.cache_clear()

1
# type() 动态创建类
2
MyClass = type('MyClass', (object,), {
3
    'x': 1,
4
    'greet': lambda self: f"Hello, I'm {self.name}"
5
})
6

7
# metaclass 控制类创建
8
class ModelMeta(type):
9
    _registry = {}
10

11
    def __new__(mcs, name, bases, attrs):
12
        cls = super().__new__(mcs, name, bases, attrs)
13
        mcs._registry[name] = cls
14
        return cls
15

16
class User(metaclass=ModelMeta):
17
    pass
18

19
print(ModelMeta._registry)  # {'User': <class '__main__.User'>}

1
class Gateway:
2
    def __getattr__(self, name):
3
        # 属性不存在时调用
4
        return f"calling {name}"
5

6
    def __getattribute__(self, name):
7
        # 每次访问属性都调用
8
        return f"intercepted {name}"
9

10
class Lazy:
11
    def __init__(self):
12
        self._data = {}
13

14
    def __getattr__(self, name):
15
        print(f"Loading {name}")
16
        return self._data.get(name)
17

18
    def __setattr__(self, name, value):
19
        if name == '_data':
20
            super().__setattr__(name, value)
21
        else:
22
            self._data[name] = value

1
# 描述器：实现 __get__, __set__, __delete__
2
class Property:
3
    def __init__(self, fget, fset=None):
4
        self.fget = fget
5
        self.fset = fset
6

7
    def __get__(self, obj, objtype=None):
8
        return self.fget(obj)
9

10
    def __set__(self, obj, value):
11
        if self.fset is None:
12
            raise AttributeError
13
        self.fset(obj, value)
14

15
class MyClass:
16
    @Property
17
    def value(self):
18
        return self._value
19

20
    @value.setter
21
    def value(self, v):
22
        self._value = v