为什么高级 Python 很重要

Python 的简洁性使其成为初学者最流行的语言，但其高级特性使其同样适合构建生产系统、数据管道和高性能 API。现代 Python（3.10+）包含媲美 TypeScript 的类型系统、类似 Rust 的结构化模式匹配，以及能处理数千并发连接的异步能力。

本指南通过 13 个高级主题和生产就绪的代码示例进行讲解。无论你是在构建 FastAPI 服务、数据工程管道还是 CLI 工具，掌握这些特性将使你的代码更易维护、更高性能、更正确。

1. 类型提示与泛型

Python 的类型系统自 PEP 484 以来发展显著。现代类型提示支持带 TypeVar 的泛型、带 ParamSpec 的可调用签名、带 Protocol 的结构化子类型和自引用类型。这些注解由 mypy、pyright 和 ruff 等工具在开发时检查，零运行时开销。

TypeVar 与泛型约束

TypeVar 创建保持类型关系的泛型类型变量。ParamSpec（PEP 612）捕获函数参数签名用于装饰器类型标注。Protocol（PEP 544）实现结构化子类型，任何具有匹配方法的类都满足协议，无需显式继承。

from typing import TypeVar, Generic, Protocol, ParamSpec, Callable
from collections.abc import Sequence

# Basic TypeVar with constraints
T = TypeVar("T")
S = TypeVar("S", bound=str)  # Upper bound: must be str or subclass
Num = TypeVar("Num", int, float)  # Value restriction: only int or float

def first(items: Sequence[T]) -> T:
    """Return the first item, preserving the exact type."""
    return items[0]

reveal_type(first([1, 2, 3]))       # int
reveal_type(first(["a", "b"]))      # str
reveal_type(first([(1, 2), (3,)]))  # tuple[int, ...]

# Generic class with TypeVar
class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []

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

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

    def peek(self) -> T:
        return self._items[-1]

int_stack: Stack[int] = Stack()
int_stack.push(42)       # OK
# int_stack.push("oops") # Type error!

ParamSpec 与 Protocol

Python 3.12 引入了新的类型参数语法（PEP 695），用更简洁的 [T] 语法简化泛型定义，替代 TypeVar 声明。

# ParamSpec — preserve function signatures in decorators
P = ParamSpec("P")
R = TypeVar("R")

def logged(func: Callable[P, R]) -> Callable[P, R]:
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        print(f"Calling {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@logged
def add(x: int, y: int) -> int:
    return x + y

add(1, 2)      # OK — type checker knows signature
# add("a", "b") # Type error: expected int

# Protocol — structural subtyping (duck typing with types)
class Renderable(Protocol):
    def render(self) -> str: ...

class Button:
    def render(self) -> str:
        return "<button>Click</button>"

class Chart:
    def render(self) -> str:
        return "<svg>...</svg>"

def display(widget: Renderable) -> None:
    print(widget.render())

display(Button())  # OK — Button has render() -> str
display(Chart())   # OK — Chart has render() -> str

# Python 3.12+ new syntax (PEP 695)
def first_new[T](items: Sequence[T]) -> T:
    return items[0]

2. Dataclass 与 Pydantic 模型

Dataclass（PEP 557）通过自动生成 __init__、__repr__、__eq__ 等方法消除数据类的样板代码。它们支持默认值、字段工厂、初始化后处理、slots 和冻结（不可变）实例。

Dataclass 高级用法

from dataclasses import dataclass, field, asdict, astuple
from typing import Optional
from datetime import datetime

@dataclass(frozen=True, slots=True)  # Immutable + memory efficient
class Point:
    x: float
    y: float

    @property
    def distance(self) -> float:
        return (self.x ** 2 + self.y ** 2) ** 0.5

@dataclass
class User:
    name: str
    email: str
    age: int
    tags: list[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    _password_hash: str = field(default="", repr=False, compare=False)

    def __post_init__(self) -> None:
        """Validate after __init__ runs."""
        if self.age < 0:
            raise ValueError("Age must be non-negative")
        if "@" not in self.email:
            raise ValueError("Invalid email")

user = User("Alice", "alice@example.com", 30, ["admin"])
print(asdict(user))    # Convert to dict
print(astuple(user))   # Convert to tuple

Pydantic v2 模型验证

Pydantic v2 通过运行时验证、序列化、JSON Schema 生成和设置管理进一步扩展数据建模。基于 Rust 核心（pydantic-core），v2 比 v1 快 5-50 倍，是 FastAPI 的基础。

from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic import ConfigDict, EmailStr
from datetime import datetime

class Address(BaseModel):
    street: str
    city: str
    country: str = "US"
    zip_code: str = Field(pattern=r"^\d{5}(-\d{4})?$")

class UserCreate(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True)

    name: str = Field(min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(ge=0, le=150)
    address: Address
    tags: list[str] = Field(default_factory=list, max_length=10)

    @field_validator("name")
    @classmethod
    def name_must_be_title_case(cls, v: str) -> str:
        return v.title()

    @model_validator(mode="after")
    def check_age_for_minors(self) -> "UserCreate":
        if self.age < 18 and "minor" not in self.tags:
            self.tags.append("minor")
        return self

# Automatic validation + serialization
user = UserCreate(
    name="  alice smith  ",
    email="alice@example.com",
    age=25,
    address={"street": "123 Main St", "city": "NYC", "zip_code": "10001"},
)
print(user.model_dump_json(indent=2))  # JSON serialization
print(user.model_json_schema())          # JSON Schema generation

简单内部数据结构使用 dataclass，外部边界（API 请求/响应体、配置文件、数据库记录、事件 schema）需要验证时使用 Pydantic。

3. 装饰器深入探讨

装饰器是 Python 最强大的元编程工具。它们使用 @decorator 语法在定义时修改函数或类。除了简单包装器外，Python 支持参数化装饰器、类装饰器、装饰器堆叠和 functools.wraps 保留元数据。

参数化装饰器

装饰器本质上是一个接受函数并返回函数的可调用对象。参数化装饰器添加一个外层函数来返回实际的装饰器。类装饰器接受一个类并返回修改后的类，实现单例、注册和自动序列化等模式。

import functools
import time
from typing import Callable, TypeVar, ParamSpec

P = ParamSpec("P")
R = TypeVar("R")

# Simple decorator with functools.wraps
def timer(func: Callable[P, R]) -> Callable[P, R]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

# Parameterized decorator (decorator factory)
def retry(max_attempts: int = 3, delay: float = 1.0):
    """Retry a function on failure with exponential backoff."""
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            last_exception: Exception | None = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    wait = delay * (2 ** attempt)
                    print(f"Attempt {attempt + 1} failed, retrying in {wait}s")
                    time.sleep(wait)
            raise last_exception  # type: ignore
        return wrapper
    return decorator

@retry(max_attempts=5, delay=0.5)
@timer
def fetch_data(url: str) -> dict:
    """Fetch data from API with retry logic."""
    import urllib.request
    import json
    with urllib.request.urlopen(url) as resp:
        return json.loads(resp.read())

类装饰器

始终在包装函数上使用 functools.wraps 以保留原始函数的 __name__、__doc__ 和 __module__。否则调试和文档工具会显示包装器的元数据。

# Class decorator: auto-register all subclasses
_registry: dict[str, type] = {}

def register(cls: type) -> type:
    """Register a class in the global registry."""
    _registry[cls.__name__] = cls
    return cls

@register
class JSONParser:
    def parse(self, data: str) -> dict:
        import json
        return json.loads(data)

@register
class XMLParser:
    def parse(self, data: str) -> dict:
        # XML parsing logic
        return {}

print(_registry)  # {"JSONParser": <class>, "XMLParser": <class>}

# Class decorator: singleton pattern
def singleton(cls: type) -> type:
    instances: dict[type, object] = {}
    @functools.wraps(cls, updated=())
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance  # type: ignore

@singleton
class DatabaseConnection:
    def __init__(self, url: str) -> None:
        self.url = url
        print(f"Connecting to {url}")

db1 = DatabaseConnection("postgres://localhost/mydb")
db2 = DatabaseConnection("postgres://localhost/mydb")
print(db1 is db2)  # True — same instance

4. 上下文管理器与生成器

上下文管理器通过 with 语句处理资源生命周期，保证清理。它们实现 __enter__ 和 __exit__（异步版本为 __aenter__/__aexit__）。contextlib 模块提供 @contextmanager、suppress() 和 ExitStack 等快捷方式。

上下文管理器模式

from contextlib import contextmanager, asynccontextmanager, ExitStack
from typing import Generator, AsyncGenerator
import time

# Class-based context manager
class Timer:
    def __init__(self, label: str) -> None:
        self.label = label
        self.elapsed: float = 0.0

    def __enter__(self) -> "Timer":
        self.start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        self.elapsed = time.perf_counter() - self.start
        print(f"{self.label}: {self.elapsed:.4f}s")
        return False  # Do not suppress exceptions

with Timer("data processing") as t:
    data = [i ** 2 for i in range(1_000_000)]

# Generator-based context manager (simpler)
@contextmanager
def temp_directory() -> Generator[str, None, None]:
    import tempfile, shutil
    path = tempfile.mkdtemp()
    try:
        yield path  # Provide the resource
    finally:
        shutil.rmtree(path)  # Guaranteed cleanup

with temp_directory() as tmpdir:
    print(f"Working in {tmpdir}")

# ExitStack — compose multiple context managers dynamically
def process_files(paths: list[str]) -> list[str]:
    with ExitStack() as stack:
        files = [stack.enter_context(open(p)) for p in paths]
        return [f.read() for f in files]

生成器与 yield from

生成器通过 yield 惰性产生值，实现对大数据集的内存高效迭代。生成器表达式提供内联语法。yield from 语法（PEP 380）委托给子生成器，实现干净的递归生成和协程组合。

from typing import Generator, Iterator

# Generator for memory-efficient processing
def read_large_file(path: str, chunk_size: int = 8192) -> Generator[str, None, None]:
    """Read a large file in chunks without loading it all."""
    with open(path) as f:
        while chunk := f.read(chunk_size):
            yield chunk

# yield from — delegate to sub-generator
def flatten(nested: list) -> Generator:
    """Recursively flatten nested lists."""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)  # Delegate recursion
        else:
            yield item

data = [1, [2, 3], [4, [5, 6]], 7]
print(list(flatten(data)))  # [1, 2, 3, 4, 5, 6, 7]

# Generator pipeline — compose data transformations
def lines(path: str) -> Iterator[str]:
    with open(path) as f:
        yield from f

def strip(it: Iterator[str]) -> Iterator[str]:
    for line in it:
        yield line.strip()

def non_empty(it: Iterator[str]) -> Iterator[str]:
    for line in it:
        if line:
            yield line

# Compose: file -> strip -> non_empty -> process
# pipeline = non_empty(strip(lines("data.txt")))
# for line in pipeline:
#     process(line)

异步上下文管理器

异步上下文管理器结合 async/await 和资源管理，对异步代码中的数据库连接、HTTP 会话和文件 I/O 至关重要。

import asyncio

@asynccontextmanager
async def db_transaction(conn) -> AsyncGenerator:
    """Async context manager for database transactions."""
    tx = await conn.begin()
    try:
        yield tx
        await tx.commit()
    except Exception:
        await tx.rollback()
        raise

# Usage:
# async with db_transaction(conn) as tx:
#     await tx.execute("INSERT INTO users ...")

5. Async/Await 模式

asyncio 是 Python 内置的 I/O 密集型并发操作框架。单线程使用协作式多任务处理数千个连接。async/await 语法使异步代码几乎和同步代码一样可读。

asyncio.gather 与信号量

asyncio.gather 并发运行多个协程并收集结果。信号量限制对共享资源的并发访问。Task group（Python 3.11+，PEP 654）提供失败时自动取消的结构化并发。

import asyncio
import httpx

async def fetch_url(client: httpx.AsyncClient, url: str) -> dict:
    """Fetch a single URL."""
    response = await client.get(url)
    return {"url": url, "status": response.status_code}

async def fetch_all(urls: list[str]) -> list[dict]:
    """Fetch multiple URLs concurrently."""
    async with httpx.AsyncClient() as client:
        tasks = [fetch_url(client, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in results if isinstance(r, dict)]

# Semaphore — limit concurrent connections
async def fetch_with_limit(urls: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_fetch(client: httpx.AsyncClient, url: str):
        async with semaphore:  # At most max_concurrent at once
            return await fetch_url(client, url)

    async with httpx.AsyncClient() as client:
        tasks = [limited_fetch(client, url) for url in urls]
        return await asyncio.gather(*tasks)

Task Group（Python 3.11+）

生产异步代码使用 asyncio.TaskGroup 实现结构化并发，httpx 用于 HTTP 请求，asyncpg 用于 PostgreSQL，motor 用于 MongoDB。始终正确处理取消并对关键操作使用 asyncio.shield。

# TaskGroup — structured concurrency (Python 3.11+)
async def process_batch(items: list[str]) -> list[str]:
    results: list[str] = []

    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_item(item))

    # All tasks completed successfully here
    # If ANY task raises, ALL are cancelled automatically
    return results

async def process_item(item: str) -> str:
    await asyncio.sleep(0.1)  # Simulate I/O
    return f"processed: {item}"

# Producer-consumer with asyncio.Queue
async def producer(queue: asyncio.Queue[str], items: list[str]):
    for item in items:
        await queue.put(item)
    await queue.put("")  # Sentinel to signal done

async def consumer(queue: asyncio.Queue[str], name: str):
    while True:
        item = await queue.get()
        if item == "":  # Sentinel
            await queue.put("")  # Pass sentinel to other consumers
            break
        print(f"{name} processing: {item}")
        await asyncio.sleep(0.1)
        queue.task_done()

async def main():
    queue: asyncio.Queue[str] = asyncio.Queue(maxsize=100)
    items = [f"item-{i}" for i in range(50)]

    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue, items))
        for i in range(3):  # 3 consumers
            tg.create_task(consumer(queue, f"worker-{i}"))

6. 元类与描述符

元类是类的类。定义类时，Python 使用其元类（通常是 type）来创建它。自定义元类拦截类创建以添加验证、注册类、修改属性或强制编码规范。

自定义元类

# Metaclass that validates class attributes
class ValidatedMeta(type):
    def __new__(mcs, name: str, bases: tuple, namespace: dict):
        # Ensure all public methods have docstrings
        for attr_name, attr_value in namespace.items():
            if callable(attr_value) and not attr_name.startswith("_"):
                if not attr_value.__doc__:
                    raise TypeError(
                        f"{name}.{attr_name} must have a docstring"
                    )
        return super().__new__(mcs, name, bases, namespace)

class APIHandler(metaclass=ValidatedMeta):
    def get(self, request):
        """Handle GET request."""
        pass

    def post(self, request):
        """Handle POST request."""
        pass

# This would raise TypeError:
# class BadHandler(metaclass=ValidatedMeta):
#     def get(self, request):  # No docstring!
#         pass

描述符与 __set_name__

描述符实现 __get__、__set__ 和 __delete__ 来控制实例上的属性访问。它们驱动 Python 的 property、staticmethod、classmethod 和 ORM 字段。__set_name__ 钩子（PEP 487）让描述符自动获知其属性名。

# Descriptor for validated attributes
class Validated:
    def __init__(self, *, min_val: float = float("-inf"), max_val: float = float("inf")):
        self.min_val = min_val
        self.max_val = max_val

    def __set_name__(self, owner: type, name: str) -> None:
        """Called automatically when class is created."""
        self.public_name = name
        self.private_name = f"_{name}"

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.private_name, None)

    def __set__(self, obj, value: float) -> None:
        if not isinstance(value, (int, float)):
            raise TypeError(f"{self.public_name} must be a number")
        if not (self.min_val <= value <= self.max_val):
            raise ValueError(
                f"{self.public_name} must be between "
                f"{self.min_val} and {self.max_val}"
            )
        setattr(obj, self.private_name, value)

class Product:
    price = Validated(min_val=0, max_val=10_000)
    quantity = Validated(min_val=0, max_val=1_000_000)

    def __init__(self, name: str, price: float, quantity: int):
        self.name = name
        self.price = price        # Triggers Validated.__set__
        self.quantity = quantity   # Triggers Validated.__set__

p = Product("Widget", 29.99, 100)  # OK
# Product("Bad", -5, 10)           # ValueError!

# __init_subclass__ — simpler alternative to metaclasses
class PluginBase:
    _plugins: dict[str, type] = {}

    def __init_subclass__(cls, *, plugin_name: str = "", **kwargs):
        super().__init_subclass__(**kwargs)
        name = plugin_name or cls.__name__.lower()
        PluginBase._plugins[name] = cls

class JSONPlugin(PluginBase, plugin_name="json"):
    pass

class YAMLPlugin(PluginBase, plugin_name="yaml"):
    pass

print(PluginBase._plugins)  # {"json": <class>, "yaml": <class>}

7. 模式匹配

结构化模式匹配（PEP 634，Python 3.10+）带来按数据形状解构和匹配的 match/case 语句。与其他语言的 switch 不同，Python 的 match 可处理序列、映射、类实例和嵌套结构。

守卫子句（if 条件）为模式 case 添加运行时检查。捕获模式将匹配值绑定到名称。或模式（|）匹配多个备选项。通配符（_）匹配任何内容但不绑定。

结构化模式匹配示例

from dataclasses import dataclass

# Matching sequences and mappings
def process_command(command: list[str]) -> str:
    match command:
        case ["quit" | "exit"]:
            return "Goodbye!"
        case ["hello", name]:
            return f"Hello, {name}!"
        case ["add", *numbers] if all(n.isdigit() for n in numbers):
            total = sum(int(n) for n in numbers)
            return f"Sum: {total}"
        case ["set", key, value]:
            return f"Setting {key} = {value}"
        case _:
            return "Unknown command"

print(process_command(["hello", "Alice"]))  # Hello, Alice!
print(process_command(["add", "1", "2", "3"]))  # Sum: 6

# Matching class instances
@dataclass
class Point:
    x: float
    y: float

@dataclass
class Circle:
    center: Point
    radius: float

@dataclass
class Rectangle:
    top_left: Point
    width: float
    height: float

def describe_shape(shape) -> str:
    match shape:
        case Circle(center=Point(x=0, y=0), radius=r):
            return f"Circle at origin with radius {r}"
        case Circle(center=Point(x=x, y=y), radius=r) if r > 100:
            return f"Large circle at ({x}, {y})"
        case Rectangle(width=w, height=h) if w == h:
            return f"Square with side {w}"
        case Rectangle(width=w, height=h):
            return f"Rectangle {w}x{h}"
        case _:
            return "Unknown shape"

匹配映射和 API 响应

模式匹配擅长解析命令结构、处理不同形状的 API 响应、处理 AST 节点和实现状态机。

# Matching dict-like structures (API responses)
def handle_response(response: dict) -> str:
    match response:
        case {"status": "ok", "data": {"users": [first, *rest]}}:
            return f"Found {1 + len(rest)} users, first: {first}"
        case {"status": "ok", "data": data}:
            return f"Success with data: {data}"
        case {"status": "error", "code": code, "message": msg}:
            return f"Error {code}: {msg}"
        case {"status": "error", **rest}:
            return f"Error with details: {rest}"
        case _:
            return "Unexpected response format"

print(handle_response({
    "status": "ok",
    "data": {"users": ["Alice", "Bob", "Charlie"]}
}))  # Found 3 users, first: Alice

print(handle_response({
    "status": "error",
    "code": 404,
    "message": "Not found"
}))  # Error 404: Not found

8. 内存管理

__slots__ 将实例属性限制为固定集合，用更紧凑的表示替代每实例的 __dict__。这减少 40-60% 的内存使用并加速属性访问。对有数百万实例的类至关重要。

__slots__ 与内存优化

import sys
import weakref
import gc

# Without __slots__: each instance has a __dict__
class PointRegular:
    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# With __slots__: fixed attribute storage, no __dict__
class PointSlots:
    __slots__ = ("x", "y")

    def __init__(self, x: float, y: float):
        self.x = x
        self.y = y

# Memory comparison
regular = PointRegular(1.0, 2.0)
slotted = PointSlots(1.0, 2.0)

print(f"Regular: {sys.getsizeof(regular)} + {sys.getsizeof(regular.__dict__)} bytes")
print(f"Slotted: {sys.getsizeof(slotted)} bytes (no __dict__)")
# Regular: ~56 + ~104 = 160 bytes
# Slotted: ~56 bytes (60% less memory!)

# With 1 million instances:
# Regular: ~160 MB
# Slotted: ~56 MB

Weakref 与垃圾回收

weakref 创建不阻止垃圾回收的引用，对缓存、观察者模式和避免循环引用泄漏至关重要。gc 模块提供对垃圾回收器的控制，包括调试引用循环。

# weakref — references that do not prevent GC
class ExpensiveObject:
    def __init__(self, name: str):
        self.name = name
    def __del__(self):
        print(f"Deleting {self.name}")

# WeakValueDictionary for caching
cache: weakref.WeakValueDictionary[str, ExpensiveObject] = (
    weakref.WeakValueDictionary()
)

obj = ExpensiveObject("data-1")
cache["data-1"] = obj

print("data-1" in cache)  # True
del obj                    # No more strong references
gc.collect()               # Force garbage collection
print("data-1" in cache)  # False — obj was collected

# Memory profiling with tracemalloc
import tracemalloc

tracemalloc.start()

# ... your code here ...
data = [dict(x=i, y=i**2) for i in range(100_000)]

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")
for stat in top_stats[:5]:
    print(stat)

内存分析使用 tracemalloc（内置）、memory_profiler 或 objgraph。跟踪分配、发现内存泄漏并优化数据结构以减少应用程序的内存占用。

9. 并发：Threading vs Multiprocessing vs Asyncio

Python 提供三种并发模型，各适合不同的工作负载。Threading 使用共享内存的 OS 线程，但 CPU 密集型工作受 GIL 限制。Multiprocessing 使用独立进程实现完全 CPU 并行但内存开销更高。Asyncio 使用单线程事件循环实现 I/O 密集型并发。

并发模式对比代码

I/O 密集型任务（HTTP 请求、数据库查询、文件操作）使用 asyncio。CPU 密集型任务（数据处理、图像处理、ML 训练）使用 multiprocessing。当无法使用 async 或与释放 GIL 的 C 库集成时使用 threading。

import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

# I/O-bound: asyncio wins
async def async_fetch_all(urls: list[str]) -> list:
    async def fetch(url: str):
        await asyncio.sleep(0.1)  # Simulate network I/O
        return url
    return await asyncio.gather(*[fetch(u) for u in urls])

# I/O-bound: threading alternative
def threaded_fetch_all(urls: list[str]) -> list:
    def fetch(url: str) -> str:
        time.sleep(0.1)  # Simulate network I/O
        return url
    with ThreadPoolExecutor(max_workers=20) as executor:
        return list(executor.map(fetch, urls))

# CPU-bound: multiprocessing wins
def cpu_task(n: int) -> int:
    """CPU-intensive computation."""
    return sum(i * i for i in range(n))

def parallel_compute(numbers: list[int]) -> list[int]:
    with ProcessPoolExecutor() as executor:
        return list(executor.map(cpu_task, numbers))

# Benchmark results (1000 URLs / 10 CPU tasks):
# asyncio:         ~0.1s (1000 concurrent coroutines)
# threading (20):  ~5.0s (20 threads, batched)
# multiprocessing: ~1.5s (CPU-bound, 8 cores)

Python 3.13 引入了自由线程构建（PEP 703），移除 GIL，实现真正的线程并行。这仍是实验性的，但代表了 Python 并发的未来。

10. 使用 pytest 测试

pytest 是 Python 测试的事实标准。它提供比 unittest 更简单的语法、用于 setup/teardown 的强大 fixture、用于数据驱动测试的 parametrize 和丰富的插件生态。

Fixture 与参数化

Fixture 管理测试依赖和状态。它们支持作用域（function、class、module、session）和使用 yield 的自动清理。conftest.py 文件无需导入即可在测试模块间共享 fixture。

import pytest
from unittest.mock import AsyncMock, patch, MagicMock

# conftest.py — shared fixtures
@pytest.fixture
def sample_user() -> dict:
    return {"name": "Alice", "email": "alice@test.com", "age": 30}

@pytest.fixture
def db_session():
    """Create a test database session with rollback."""
    session = create_test_session()
    yield session          # Provide to test
    session.rollback()     # Cleanup after test
    session.close()

@pytest.fixture(scope="module")
def api_client():
    """Shared HTTP client for the entire test module."""
    client = TestClient(app)
    yield client
    client.close()

# Parametrize — generate multiple test cases
@pytest.mark.parametrize("input_val, expected", [
    ("hello", "HELLO"),
    ("World", "WORLD"),
    ("", ""),
    ("123abc", "123ABC"),
    ("already UPPER", "ALREADY UPPER"),
])
def test_uppercase(input_val: str, expected: str):
    assert input_val.upper() == expected

# Parametrize with IDs for clear test names
@pytest.mark.parametrize("a, b, expected", [
    pytest.param(1, 2, 3, id="positive"),
    pytest.param(-1, 1, 0, id="negative-positive"),
    pytest.param(0, 0, 0, id="zeros"),
])
def test_add(a: int, b: int, expected: int):
    assert a + b == expected

Mocking 与异步测试

使用 unittest.mock 或 pytest-mock 隔离被测单元。Parametrize 从数据生成多个测试用例。Marker（@pytest.mark）对测试分类以选择性执行。

# Mocking external services
class UserService:
    def __init__(self, api_client):
        self.api_client = api_client

    async def get_user(self, user_id: int) -> dict:
        response = await self.api_client.get(f"/users/{user_id}")
        return response.json()

def test_get_user_with_mock():
    mock_client = MagicMock()
    mock_client.get.return_value.json.return_value = {
        "id": 1, "name": "Alice"
    }
    service = UserService(mock_client)
    # ... test logic

# Async test (pytest-asyncio)
@pytest.mark.asyncio
async def test_async_fetch():
    mock_client = AsyncMock()
    mock_client.get.return_value.json.return_value = {"status": "ok"}

    service = UserService(mock_client)
    result = await service.get_user(1)
    assert result == {"status": "ok"}
    mock_client.get.assert_called_once_with("/users/1")

# Patching module-level dependencies
@patch("myapp.services.requests.get")
def test_external_api(mock_get):
    mock_get.return_value.status_code = 200
    mock_get.return_value.json.return_value = {"data": [1, 2, 3]}
    # ... test your function that calls requests.get

# Custom markers for test categorization
@pytest.mark.slow
def test_large_dataset_processing():
    """Run with: pytest -m slow"""
    pass

@pytest.mark.integration
def test_database_connection():
    """Run with: pytest -m integration"""
    pass

11. Python 打包

现代 Python 打包以 pyproject.toml（PEP 621）为中心，替代 setup.py 和 setup.cfg。它在一个文件中定义项目元数据、依赖、构建系统和工具配置。

pyproject.toml

# pyproject.toml — modern Python project configuration
[project]
name = "my-awesome-lib"
version = "1.0.0"
description = "A high-performance data processing library"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.11"
authors = [{name = "Alice", email = "alice@example.com"}]
keywords = ["data", "processing", "etl"]
classifiers = [
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Programming Language :: Python :: 3.13",
]

dependencies = [
    "httpx>=0.27",
    "pydantic>=2.0",
    "rich>=13.0",
]

[project.optional-dependencies]
dev = ["pytest>=8.0", "ruff>=0.8", "mypy>=1.13"]
docs = ["mkdocs-material>=9.0"]

[project.scripts]
my-cli = "my_lib.cli:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.ruff]
line-length = 88
target-version = "py311"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]

[tool.mypy]
python_version = "3.11"
strict = true

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

uv：现代 Python 包管理

构建后端包括 setuptools（标准）、Hatch（现代）、Poetry（依赖管理+发布）和 PDM（PEP 582）。uv（来自 Astral，ruff 的制作者）是最快的包安装器和解析器，比 pip 快 10-100 倍。

# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create a new project
uv init my-project
cd my-project

# Add dependencies (resolves and installs in seconds)
uv add httpx pydantic rich
uv add --dev pytest ruff mypy

# Run scripts (auto-creates venv if needed)
uv run python main.py
uv run pytest
uv run ruff check .

# Pin Python version
uv python pin 3.12

# Lock dependencies for reproducibility
uv lock

# Build and publish
uv build
uv publish

# Run a one-off script with inline dependencies
uv run --with requests --with rich script.py

# Speed comparison (adding 10 packages):
# pip:    45.2s
# poetry: 32.1s
# uv:     0.4s  (100x faster!)

2026 年新项目使用 uv 管理依赖：uv init 创建项目，uv add 添加依赖，uv run 执行脚本，uv publish 上传到 PyPI。它自动处理虚拟环境。

12. 性能优化

从分析开始优化：cProfile 用于函数级计时，line_profiler 用于逐行分析，py-spy 用于生产代码的采样式分析。永远不要在没有分析的情况下优化。

分析与记忆化

import cProfile
import functools
import time

# cProfile — function-level profiling
def profile_me():
    data = [i ** 2 for i in range(1_000_000)]
    sorted_data = sorted(data, reverse=True)
    return sum(sorted_data[:100])

cProfile.run("profile_me()", sort="cumulative")
# Output:
#   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
#        1    0.000    0.000    0.412    0.412 <string>:1(<module>)
#        1    0.231    0.231    0.412    0.412 script.py:5(profile_me)
#        1    0.181    0.181    0.181    0.181 {built-in method builtins.sorted}

# functools.lru_cache — memoization
@functools.lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# Without cache: fibonacci(35) takes ~5 seconds
# With cache: fibonacci(35) takes ~0.00001 seconds
print(fibonacci(100))  # Instant!
print(fibonacci.cache_info())  # Hits, misses, size

# functools.cache — unlimited cache (Python 3.9+)
@functools.cache
def expensive_computation(x: int, y: int) -> float:
    time.sleep(1)  # Simulate expensive work
    return (x ** y) / (x + y)

NumPy 向量化 vs Python 循环

functools.lru_cache 和 functools.cache 的记忆化消除冗余计算。NumPy 向量化用 C 级操作替代 Python 循环，数值数据获得 10-100 倍加速。

import numpy as np
import time

# Python loop: slow
def python_distance(x1, y1, x2, y2):
    """Calculate distances using pure Python."""
    distances = []
    for i in range(len(x1)):
        d = ((x1[i] - x2[i])**2 + (y1[i] - y2[i])**2) ** 0.5
        distances.append(d)
    return distances

# NumPy vectorized: fast
def numpy_distance(x1, y1, x2, y2):
    """Calculate distances using NumPy vectorization."""
    return np.sqrt((x1 - x2)**2 + (y1 - y2)**2)

# Benchmark with 1 million points
n = 1_000_000
x1 = np.random.rand(n)
y1 = np.random.rand(n)
x2 = np.random.rand(n)
y2 = np.random.rand(n)

start = time.perf_counter()
python_distance(list(x1), list(y1), list(x2), list(y2))
python_time = time.perf_counter() - start

start = time.perf_counter()
numpy_distance(x1, y1, x2, y2)
numpy_time = time.perf_counter() - start

print(f"Python: {python_time:.3f}s")
print(f"NumPy:  {numpy_time:.3f}s")
print(f"Speedup: {python_time / numpy_time:.0f}x")
# Python: 1.234s
# NumPy:  0.012s
# Speedup: ~100x

Cython 与编译加速

要获得最高性能，Cython 将 Python 编译为带可选静态类型的 C。替代方案包括 Numba（数值代码 JIT）、mypyc（编译类型注解的 Python）和 PyO3（用 Rust 编写 Python 扩展）。

# math_ops.pyx — Cython source file
# cython: language_level=3

def primes_python(int limit):
    """Find primes up to limit using Sieve of Eratosthenes."""
    cdef int i, j
    cdef list sieve = [True] * (limit + 1)
    cdef list result = []

    for i in range(2, limit + 1):
        if sieve[i]:
            result.append(i)
            for j in range(i * i, limit + 1, i):
                sieve[j] = False
    return result

# setup.py for Cython
# from setuptools import setup
# from Cython.Build import cythonize
# setup(ext_modules=cythonize("math_ops.pyx"))

# Alternative: Numba JIT (no Cython setup needed)
# from numba import njit
# @njit
# def fast_sum(arr):
#     total = 0.0
#     for val in arr:
#         total += val
#     return total

13. Python 中的设计模式

经典设计模式在 Python 中因一等函数、鸭子类型和动态特性而不同。许多在 Java 中需要复杂类层次结构的 GoF 模式在 Python 中变成简单的函数或装饰器。

单例与工厂模式

# Singleton — using module-level instance (Pythonic way)
# config.py
class _Config:
    def __init__(self):
        self._settings: dict = {}

    def get(self, key: str, default=None):
        return self._settings.get(key, default)

    def set(self, key: str, value) -> None:
        self._settings[key] = value

config = _Config()  # Module-level singleton
# Usage: from config import config

# Factory — using a registry dictionary
from typing import Protocol

class Serializer(Protocol):
    def serialize(self, data: dict) -> str: ...
    def deserialize(self, raw: str) -> dict: ...

class JSONSerializer:
    def serialize(self, data: dict) -> str:
        import json
        return json.dumps(data)
    def deserialize(self, raw: str) -> dict:
        import json
        return json.loads(raw)

class YAMLSerializer:
    def serialize(self, data: dict) -> str:
        import yaml
        return yaml.dump(data)
    def deserialize(self, raw: str) -> dict:
        import yaml
        return yaml.safe_load(raw)

_serializers: dict[str, type[Serializer]] = {
    "json": JSONSerializer,
    "yaml": YAMLSerializer,
}

def get_serializer(format: str) -> Serializer:
    """Factory function to create serializers."""
    cls = _serializers.get(format)
    if cls is None:
        raise ValueError(f"Unknown format: {format}")
    return cls()

观察者与策略模式

单例模式使用模块级实例或 __new__。工厂模式利用可调用对象的字典。观察者模式使用 weakref 集合。策略模式直接传递函数。Python 的动态特性使许多模式比 Java 对应物更轻量。

import weakref
from typing import Callable

# Observer pattern with weakref
class EventEmitter:
    def __init__(self):
        self._listeners: dict[str, list[weakref.ref]] = {}

    def on(self, event: str, callback: Callable) -> None:
        if event not in self._listeners:
            self._listeners[event] = []
        self._listeners[event].append(weakref.ref(callback))

    def emit(self, event: str, *args, **kwargs) -> None:
        if event not in self._listeners:
            return
        alive = []
        for ref in self._listeners[event]:
            callback = ref()
            if callback is not None:
                callback(*args, **kwargs)
                alive.append(ref)
        self._listeners[event] = alive  # Prune dead refs

# Strategy pattern — functions as strategies
from typing import Callable

SortStrategy = Callable[[list], list]

def bubble_sort(data: list) -> list:
    arr = data.copy()
    for i in range(len(arr)):
        for j in range(len(arr) - i - 1):
            if arr[j] > arr[j + 1]:
                arr[j], arr[j + 1] = arr[j + 1], arr[j]
    return arr

def quick_sort(data: list) -> list:
    if len(data) <= 1:
        return data
    pivot = data[len(data) // 2]
    left = [x for x in data if x < pivot]
    middle = [x for x in data if x == pivot]
    right = [x for x in data if x > pivot]
    return quick_sort(left) + middle + quick_sort(right)

class DataProcessor:
    def __init__(self, strategy: SortStrategy = sorted):
        self.sort = strategy  # Inject strategy

    def process(self, data: list) -> list:
        return self.sort(data)

processor = DataProcessor(strategy=quick_sort)
print(processor.process([3, 1, 4, 1, 5, 9]))  # [1, 1, 3, 4, 5, 9]

理解何时应用这些模式、何时 Python 的内置特性就足够是编写惯用、可维护代码的关键。

总结

高级 Python 特性将语言从脚本工具转变为全面的软件工程平台。类型提示在运行前捕获 bug。Async/await 处理大规模并发。模式匹配使复杂逻辑声明化。uv 和 ruff 等现代工具使开发体验快速可靠。

从与当前项目最相关的特性开始。类型提示和 dataclass 为任何代码库提供即时价值。Async/await 和测试模式对 Web 服务至关重要。元类和描述符在构建框架和库时变得重要。性能优化和设计模式完善你构建生产级 Python 应用的工具集。

FAQ