1
import functools
2

3
def timer(func):
4
    """计时装饰器：测量函数执行时间"""
5
    @functools.wraps(func)
6
    def wrapper(*args, **kwargs):
7
        import time
8
        start = time.time()
9
        result = func(*args, **kwargs)
10
        print(f"{func.__name__} took {time.time() - start:.2f}s")
11
        return result
12
    return wrapper
13

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

19
slow_function()  # 输出: slow_function took 1.00s

1
def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
2
    """重试装饰器：失败自动重试"""
3
    def decorator(func):
4
        @functools.wraps(func)
5
        def wrapper(*args, **kwargs):
6
            import time
7
            for attempt in range(max_attempts):
8
                try:
9
                    return func(*args, **kwargs)
10
                except exceptions as e:
11
                    if attempt == max_attempts - 1:
12
                        raise
13
                    print(f"Retry {attempt + 1}/{max_attempts}: {e}")
14
                    time.sleep(delay)
15
        return wrapper
16
    return decorator
17

18
# 使用默认参数
19
@retry()
20
def unstable_api_call():
21
    pass
22

23
# 自定义重试策略
24
@retry(max_attempts=5, delay=2.0, exceptions=(ConnectionError, TimeoutError))
25
def fetch_data(url):
26
    pass

1
class CountCalls:
2
    """统计函数调用次数"""
3
    def __init__(self, func):
4
        functools.update_wrapper(self, func)
5
        self.func = func
6
        self.count = 0
7

8
    def __call__(self, *args, **kwargs):
9
        self.count += 1
10
        print(f"{self.func.__name__} called {self.count} times")
11
        return self.func(*args, **kwargs)
12

13
@CountCalls
14
def say_hello(name):
15
    return f"Hello, {name}"
16

17
say_hello("Alice")  # say_hello called 1 times
18
say_hello("Bob")    # say_hello called 2 times

1
def fibonacci(n):
2
    """斐波那契生成器"""
3
    a, b = 0, 1
4
    for _ in range(n):
5
        yield a
6
        a, b = b, a + b
7

8
# 使用
9
for i in fibonacci(10):
10
    print(i)
11

12
# 生成器表达式（比列表推导更省内存）
13
squares = (x * x for x in range(1_000_000))
14
# 列表推导 [x * x for x in range(1_000_000)] 会立即分配内存

1
def read_large_file(filepath, chunk_size=8192):
2
    """按块读取大文件，不会一次性加载到内存"""
3
    with open(filepath, 'r') as f:
4
        while True:
5
            chunk = f.read(chunk_size)
6
            if not chunk:
7
                break
8
            yield chunk
9

10
def csv_rows(filepath):
11
    """逐行解析 CSV，生成字典"""
12
    import csv
13
    with open(filepath, 'r') as f:
14
        reader = csv.DictReader(f)
15
        for row in reader:
16
            yield row
17

18
# 管道式处理：生成器链
19
def filter_active(users):
20
    for user in users:
21
        if user['active']:
22
            yield user
23

24
def extract_names(users):
25
    for user in users:
26
        yield user['name']
27

28
# 串联生成器
29
active_names = extract_names(filter_active(csv_rows('users.csv')))

1
def flatten(nested):
2
    """递归展平嵌套列表"""
3
    for item in nested:
4
        if isinstance(item, (list, tuple)):
5
            yield from flatten(item)
6
        else:
7
            yield item
8

9
list(flatten([1, [2, [3, 4]], 5]))  # [1, 2, 3, 4, 5]
10

11
# yield from 也可以用于子生成器通信
12
def accumulator():
13
    total = 0
14
    while True:
15
        value = yield total
16
        if value is None:
17
            break
18
        total += value
19

20
def delegator():
21
    result = yield from accumulator()
22
    print(f"Total: {result}")
23

24
gen = delegator()
25
next(gen)           # 启动生成器
26
gen.send(10)        # 返回 10
27
gen.send(20)        # 返回 30
28
gen.send(None)      # 打印 Total: 30, 抛出 StopIteration

1
class Range:
2
    def __init__(self, start, stop=None, step=1):
3
        if stop is None:
4
            start, stop = 0, start
5
        self.current = start
6
        self.stop = stop
7
        self.step = step
8

9
    def __iter__(self):
10
        return self
11

12
    def __next__(self):
13
        if self.step > 0 and self.current >= self.stop:
14
            raise StopIteration
15
        if self.step < 0 and self.current <= self.stop:
16
            raise StopIteration
17
        value = self.current
18
        self.current += self.step
19
        return value
20

21
# 使用
22
for i in Range(5):        # 0, 1, 2, 3, 4
23
    print(i)
24
for i in Range(2, 10, 3): # 2, 5, 8
25
    print(i)

1
with open('file.txt', 'r') as f:
2
    content = f.read()
3
# 自动关闭文件，即使发生异常

1
class Timer:
2
    """计时上下文管理器"""
3
    def __enter__(self):
4
        import time
5
        self.start = time.time()
6
        return self
7

8
    def __exit__(self, exc_type, exc_val, exc_tb):
9
        import time
10
        print(f"Elapsed: {time.time() - self.start:.2f}s")
11
        return False  # 不吞掉异常
12

13
with Timer() as t:
14
    # 操作
15
    pass

1
from contextlib import contextmanager
2

3
@contextmanager
4
def timer(name="Block"):
5
    import time
6
    start = time.time()
7
    try:
8
        yield
9
    finally:
10
        print(f"{name} took {time.time() - start:.2f}s")
11

12
with timer("Database query"):
13
    # 执行数据库查询
14
    pass
15

16
# 更实用的例子：临时切换工作目录
17
@contextmanager
18
def change_dir(path):
19
    import os
20
    old_dir = os.getcwd()
21
    os.chdir(path)
22
    try:
23
        yield
24
    finally:
25
        os.chdir(old_dir)
26

27
with change_dir('/tmp'):
28
    # 在 /tmp 下执行操作
29
    pass

1
from contextlib import suppress, redirect_stdout
2
import io
3

4
# suppress：忽略指定异常
5
with suppress(FileNotFoundError):
6
    os.remove('nonexistent.txt')  # 文件不存在也不会报错
7

8
# redirect_stdout：重定向标准输出
9
f = io.StringIO()
10
with redirect_stdout(f):
11
    print("Hello")
12
output = f.getvalue()  # "Hello\n"

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

4
def process(items: list[int], config: dict[str, str]) -> tuple[bool, str]:
5
    return True, "success"
6

7
# 可选类型
8
from typing import Optional
9
def find_user(user_id: int) -> Optional[str]:
10
    # 返回 str 或 None
11
    return None

1
from typing import TypeVar, Protocol, Generic
2

3
T = TypeVar('T')
4

5
class Readable(Protocol):
6
    def read(self) -> bytes: ...
7

8
def load(reader: Readable) -> bytes:
9
    return reader.read()
10

11
# 泛型容器
12
class Stack(Generic[T]):
13
    def __init__(self) -> None:
14
        self._items: list[T] = []
15

16
    def push(self, item: T) -> None:
17
        self._items.append(item)
18

19
    def pop(self) -> T:
20
        return self._items.pop()
21

22
stack: Stack[int] = Stack()
23
stack.push(42)

1
from typing import TypedDict, Literal
2

3
# 精确定义字典结构
4
class UserInfo(TypedDict):
5
    name: str
6
    age: int
7
    email: str
8

9
def create_user(info: UserInfo) -> None:
10
    print(info['name'])
11

12
# 限制取值范围
13
Mode = Literal['read', 'write', 'execute']
14
def set_permission(mode: Mode) -> None:
15
    pass
16

17
set_permission('read')      # OK
18
set_permission('delete')    # mypy 报错

1
from typing import TypeGuard
2

3
def is_str_list(val: list[object]) -> TypeGuard[list[str]]:
4
    return all(isinstance(x, str) for x in val)
5

6
def process(items: list[object]):
7
    if is_str_list(items):
8
        # 在这个分支里，mypy 知道 items 是 list[str]
9
        print(items[0].upper())

1
from dataclasses import dataclass, field
2

3
@dataclass
4
class User:
5
    name: str
6
    email: str
7
    age: int = 0                    # 默认值
8
    tags: list[str] = field(default_factory=list)  # 可变默认值必须用 factory
9
    id: int = field(init=False)     # 不在 __init__ 中
10

11
    def __post_init__(self):
12
        # __init__ 执行后自动调用
13
        self.id = hash(self.email)
14

15
user = User(name="Alice", email="alice@example.com", age=30)
16
print(user)  # User(name='Alice', email='alice@example.com', age=30, tags=[], id=...)

1
@dataclass(frozen=True)
2
class Point:
3
    x: float
4
    y: float
5

6
p = Point(1.0, 2.0)
7
# p.x = 3.0  # 会抛出 FrozenInstanceError
8
# frozen dataclass 可以作为字典键或放入 set

1
# class MyClass 等价于
2
MyClass = type('MyClass', (), {})
3

4
# 自定义元类
5
class SingletonMeta(type):
6
    """单例元类"""
7
    _instances = {}
8

9
    def __call__(cls, *args, **kwargs):
10
        if cls not in cls._instances:
11
            cls._instances[cls] = super().__call__(*args, **kwargs)
12
        return cls._instances[cls]
13

14
class Database(metaclass=SingletonMeta):
15
    def __init__(self):
16
        self.connected = True
17

18
db1 = Database()
19
db2 = Database()
20
print(db1 is db2)  # True

1
class RegisteredBase:
2
    """自动注册子类"""
3
    _registry = {}
4

5
    def __init_subclass__(cls, **kwargs):
6
        super().__init_subclass__(**kwargs)
7
        RegisteredBase._registry[cls.__name__] = cls
8

9
class HandlerA(RegisteredBase):
10
    pass
11

12
class HandlerB(RegisteredBase):
13
    pass
14

15
print(RegisteredBase._registry)
16
# {'HandlerA': <class 'HandlerA'>, 'HandlerB': <class 'HandlerB'>}

1
import asyncio
2

3
async def fetch_data(url: str) -> str:
4
    """模拟异步 HTTP 请求"""
5
    await asyncio.sleep(1)  # 模拟 IO 等待
6
    return f"Data from {url}"
7

8
async def main():
9
    # 串行执行：总共 3 秒
10
    result1 = await fetch_data("https://api1.com")
11
    result2 = await fetch_data("https://api2.com")
12
    result3 = await fetch_data("https://api3.com")
13

14
    # 并发执行：总共 1 秒
15
    results = await asyncio.gather(
16
        fetch_data("https://api1.com"),
17
        fetch_data("https://api2.com"),
18
        fetch_data("https://api3.com"),
19
    )
20

21
asyncio.run(main())

1
async def async_read_lines(filepath):
2
    """异步逐行读取文件"""
3
    import aiofiles
4
    async with aiofiles.open(filepath, 'r') as f:
5
        async for line in f:
6
            yield line.strip()
7

8
async def process_logs():
9
    async for line in async_read_lines('/var/log/app.log'):
10
        if 'ERROR' in line:
11
            print(line)

1
import aiohttp
2

3
async def fetch_json(url):
4
    async with aiohttp.ClientSession() as session:
5
        async with session.get(url) as response:
6
            return await response.json()

1
import asyncio
2

3
async def producer(queue: asyncio.Queue):
4
    for i in range(10):
5
        await queue.put(i)
6
        print(f"Produced: {i}")
7
        await asyncio.sleep(0.1)
8

9
async def consumer(queue: asyncio.Queue, name: str):
10
    while True:
11
        item = await queue.get()
12
        print(f"Consumer {name} got: {item}")
13
        queue.task_done()
14

15
async def main():
16
    queue = asyncio.Queue(maxsize=5)
17
    # 启动生产者和消费者
18
    await asyncio.gather(
19
        producer(queue),
20
        consumer(queue, "A"),
21
        consumer(queue, "B"),
22
    )
23

24
asyncio.run(main())

1
# cProfile 使用
2
import cProfile
3
cProfile.run('main()', sort='cumulative')
4

5
# 行级分析
6
# pip install line_profiler
7
@profile
8
def hot_function():
9
    pass
10

11
# 内存分析
12
# pip install memory_profiler
13
@profile
14
def memory_heavy():
15
    big_list = [i for i in range(1000000)]
16
    return sum(big_list)

1
class Point:
2
    __slots__ = ('x', 'y')
3

4
    def __init__(self, x, y):
5
        self.x = x
6
        self.y = y
7

8
# 普通 class 每个实例有 __dict__，占用约 56 字节
9
# 使用 __slots__ 的实例约 16 字节
10
# 创建 100 万个实例可以节省约 40MB 内存
11

12
# 缺点：不能动态添加属性
13
# p = Point(1, 2)
14
# p.z = 3  # AttributeError