Python与Rust混合编程实战：高性能数据处理和Web应用后端构建等案例

# 在 Cargo.toml 中添加
[dependencies]
pyo3 = { version = "0.18", features = ["extension-module"] }
[lib]
crate-type = ["cdylib"]

use pyo3::prelude::*;
/// 计算斐波那契数列
#[pyfunction]
fn fibonacci(n: u64) -> u64 {
    match n {
        0 => 0,
        1 => 1,
        _ => fibonacci(n - 1) + fibonacci(n - 2),
    }
}
/// 一个简单的加法函数
#[pyfunction]
fna add(a: i32, b: i32) -> i32 {
    a + b
}
/// 定义 Python 模块
#[pymodule]
fn rust_extension(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(fibonacci, m)?)?;
    m.add_function(wrap_pyfunction!(add, m)?)?;
    Ok(() )
}

# 构建
cargo build --release
# 将生成的 .so 文件重命名为 Python 可导入的名称
cp target/release/librust_extension.so rust_extension.so
# 在 Python 中使用
import rust_extension
print(rust_extension.add(1, 2))  # 输出: 3
print(rust_extension.fibonacci(10))  # 输出: 55

// src/lib.rs
#[no_mangle]
pub extern "C" fn add(a: i32, b: i32) -> i32 {
    a + b
}

# Cargo.toml
[lib]
crate-type = ["cdylib"]

# 使用 ctypes 调用
import ctypes

# 加载 Rust 库
lib = ctypes.CDLL('./target/release/libmy_lib.so')

# 定义函数签名
lib.add.argtypes = [ctypes.c_int, ctypes.c_int]
lib.add.restype = ctypes.c_int

# 调用函数
result = lib.add(1, 2)
print(result)  # 输出: 3

// src/main.rs
fn main() {
    let args: Vec<String> = std::env::args().collect();
    if args.len() == 3 {
        let a: i32 = args[1].parse().unwrap();
        let b: i32 = args[2].parse().unwrap();
        println!("{}", a + b);
    }
}

# 使用 subprocess 调用
import subprocess

result = subprocess.run(['./target/release/add', '1', '2'], capture_output=True, text=True)
print(result.stdout.strip())  # 输出: 3

// src/lib.rs
use pyo3::prelude::*;
use csv::ReaderBuilder;
use std::fs::File;
#[pyfunction]
pub fn calculate_average(file_path: &str, column_index: usize) -> PyResult<f64> {
    let file = File::open(file_path).map_err(|e| PyErr::new::<pyo3::exceptions::PyIOError, _>(e.to_string()))?;
    let mut rdr = ReaderBuilder::new().has_headers(true).from_reader(file);
    let mut sum = 0.0;
    let mut count = 0;
    for result in rdr.records() {
        let record = result.map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))?;
        if let Some(value) = record.get(column_index) {
            if let Ok(num) = value.parse::<f64>() {
                sum += num;
                count += 1;
            }
        }
    }
    if count == 0 {
        return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>("No valid numbers found in column"));
    }
    Ok(sum / count as f64)
}
#[pymodule]
fn csv_processor(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(calculate_average, m)?)?;
    Ok(() )
}

# Cargo.toml
[package]
name = "csv-processor"
version = "0.1.0"
edition = "2021"
[dependencies]
pyo3 = { version = "0.18", features = ["extension-module"] }
csv = "1.2"
[lib]
crate-type = ["cdylib"]

import csv_processor

# 生成测试数据
import csv
import random

with open('test_data.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['id', 'value'])
    for i in range(1000000):
        writer.writerow([i, random.random() * 100])

# 使用 Rust 函数计算平均值
import time

start_time = time.time()
average = csv_processor.calculate_average('test_data.csv', 1)
end_time = time.time()

print(f"平均值: {average}")
print(f"处理时间: {end_time - start_time:.2f} 秒")

# 与纯 Python 实现比较
start_time = time.time()
sum_val = 0
count = 0

with open('test_data.csv', 'r') as f:
    reader = csv.reader(f)
    next(reader)  # 跳过表头
    for row in reader:
        sum_val += float(row[1])
        count += 1

average_py = sum_val / count
end_time = time.time()

print(f"Python 平均值: {average_py}")
print(f"Python 处理时间: {end_time - start_time:.2f} 秒")

// src/lib.rs
use pyo3::prelude::*;
#[pyfunction]
pub fn compute_prime_numbers(limit: u64) -> PyResult<Vec<u64>> {
    let mut primes = Vec::new();
    for num in 2..=limit {
        if is_prime(num) {
            primes.push(num);
        }
    }
    Ok(primes)
}
fn is_prime(n: u64) -> bool {
    if n <= 1 {
        return false;
    }
    for i in 2..=(n as f64).sqrt() as u64 {
        if n % i == 0 {
            return false;
        }
    }
    true
}
#[pymodule]
fn prime_calculator(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(compute_prime_numbers, m)?)?;
    Ok(() )
}

from flask import Flask, request, jsonify
import prime_calculator

app = Flask(__name__)

@app.route('/api/primes', methods=['GET'])
def get_primes():
    limit = request.args.get('limit', type=int, default=100)
    
    try:
        primes = prime_calculator.compute_prime_numbers(limit)
        return jsonify({
            'limit': limit,
            'primes': primes,
            'count': len(primes)
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(debug=True)

# 安装 maturin
pip install maturin
# 构建和安装
maturin develop