学 Rust 怎么让你写出更好的 Python

# 习惯性写法：类型注解是摆设
def get_user_name(user_id):
    # 返回 str？还是 None？调用者猜
    result = db.query("SELECT name FROM users WHERE id = ?", user_id)
    return result  # 可能是 None，但签名没说

# Rust 思维改造后：类型是契约，None 必须显式声明
from typing import Optional

def get_user_name(user_id: int) -> Optional[str]:
    """返回用户名，用户不存在时返回 None。调用方必须处理 None。"""
    result = db.query("SELECT name FROM users WHERE id = ?", user_id)
    return result

name = get_user_name(42)
if name is None:
    # Rust 的 match None 分支在 Python 里变成显式处理
    logger.warning("User 42 not found")
    return "anonymous"
print(f"Hello, {name}")

# 兜底式：出了问题只知道"出问题了"
try:
    data = fetch_api_data(url)
    parsed = parse_json(data)
    save_to_db(parsed)
except Exception as e:
    logger.error(f"Something went wrong: {e}")
    raise

# Rust 思维改造后：每个失败点有名字，处理路径清晰
class FetchError(Exception): pass
class ParseError(Exception): pass
class DBError(Exception): pass

def process_data(url: str) -> None:
    try:
        data = fetch_api_data(url)
    except FetchError as e:
        logger.error(f"API 请求失败: {e}")
        raise

    try:
        parsed = parse_json(data)
    except ParseError as e:
        logger.error(f"JSON 解析失败，原始数据: {data[:200]}")
        raise

    try:
        save_to_db(parsed)
    except DBError as e:
        logger.error(f"数据库写入失败: {e}")
        # 可以选择重试而非直接 raise
        retry_save(parsed, max_attempts=3)

# 以前：靠 GC 和运气
def process_files(paths):
    results = []
    for p in paths:
        f = open(p)           # 谁关？GC 某个时刻？还是永远不关？
        results.append(f.read())
    return results

# Rust 思维改造后：资源生命周期必须显式
from contextlib import closing
import sqlite3

def process_files(paths: list[str]) -> list[str]:
    results = []
    for p in paths:
        with open(p) as f:    # 作用域结束 = Rust 的 drop，显式释放
            results.append(f.read())
    return results

def query_user_data(db_path: str, user_id: int) -> dict:
    # 连接的生命周期绑定到这个函数，不泄漏到外部
    with closing(sqlite3.connect(db_path)) as conn:
        with closing(conn.cursor()) as cur:
            cur.execute("SELECT * FROM users WHERE id = ?", (user_id,))
            row = cur.fetchone()
            return dict(row) if row else {}

from typing import Final

# Rust 思维：标记不该变的东西
MAX_CONNECTIONS: Final[int] = 100
API_BASE_URL: Final[str] = "https://api.example.com"

# 函数参数用文档约定不可变语义
def calculate_total(items: list[dict]) -> float:
    """计算总价。不修改 items 内容——调用方可以放心传原始列表。"""
    # Rust 习惯：不就地修改，返回新值
    return sum(item["price"] * item["quantity"] for item in items)

# 对比就地修改版本（Rust 需要 &mut 才允许，Python 默认允许）
def calculate_total_mut(items: list[dict]) -> float:
    """计算总价并给每个 item 加上 _total 字段。调用方注意：items 被修改了。"""
    total = 0.0
    for item in items:
        item["_total"] = item["price"] * item["quantity"]  # 就地修改
        total += item["_total"]
    return total

# 1. 创建项目
mkdir rust_for_py && cd rust_for_py
pip install maturin

# 2. 初始化 Rust 项目
maturin init --lib
# 选择 pyo3 绑定

# 3. 编辑 src/lib.rs

use pyo3::prelude::*;

/// 计算字符串中每个字符的出现次数
/// Rust 版本：类型明确，返回值不可变，错误由编译器拦截
#[pyfunction]
fn char_counts(text: &str) -> PyResult<Vec<(char, usize)>> {
    // Rust 强制你声明类型、处理可能的错误
    let mut counts: std::collections::HashMap<char, usize> = 
        std::collections::HashMap::new();

    for ch in text.chars() {
        *counts.entry(ch).or_insert(0) += 1;
    }

    // 转成排序后的 Vec 返回给 Python
    let mut result: Vec<(char, usize)> = counts.into_iter().collect();
    result.sort_by(|a, b| b.1.cmp(&a.1));
    Ok(result)
}

/// 一个 Python 模块
#[pymodule]
fn rust_for_py(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(char_counts, m)?)?;
    Ok(())
}

# 4. 编译并开发模式安装
maturin develop

# 5. 在 Python 中使用
python -c "
from rust_for_py import char_counts
result = char_counts('hello rust, hello python')
print(result)
# [(' ', 2), ('h', 2), ('l', 4), ('e', 2), ('o', 3), ...]
"