大家好，今天为大家分享一个超实用的 Python 库 - returns。

Github地址：https://github.com/dry-python/returns

returns是由dry-python团队开发的一个强大的Python第三方库，专门设计用于函数式编程和类型安全的错误处理。该库的核心理念是让函数返回更有意义、可预测且类型安全的值，而不是依赖传统的异常处理机制。使用returns库，可以避免常见的None检查和异常处理陷阱，构建出更加优雅的业务逻辑流程，同时保持完整的类型注解支持和mypy静态类型检查兼容性。

安装

1、安装方法

returns库可以通过pip包管理器进行安装：

pip install returns

对于需要完整功能支持的用户，建议安装额外依赖：

pip install returns[full]

如果需要开发环境支持，可以安装开发版本：

pip install returns[dev]

2、验证安装

安装完成后，可以通过以下Python代码测试基本功能：

result = Success("Hello, returns!")
print(result.unwrap())  # 输出: Hello, returns!

特性

• 类型安全保证：提供完整的类型注解支持，与mypy静态类型检查器完美兼容
• 多种容器类型：包含Result、Maybe、IO、Future等多种函数式编程容器
• 异常安全处理：通过@safe装饰器将可能抛出异常的函数转换为安全的Result类型
• 链式操作支持：提供map、bind、apply等方法实现优雅的函数组合
• do-notation语法：支持类似Haskell的do-notation语法糖，简化复杂的嵌套操作
• 依赖注入机制：通供RequiresContext容器实现类型安全的依赖注入
• 异步编程支持：提供Future和FutureResult容器处理异步操作

基本功能

1、Result容器基础使用

Result容器是returns库的核心组件，用于处理可能失败的操作而不抛出异常。它只有两种状态：Success表示成功，Failure表示失败。

from returns.result import Result, Success, Failure

def divide_numbers(a: int, b: int) -> Result[float, str]:
    """安全的除法操作，返回Result类型"""
    if b == 0:
        return Failure("除数不能为零")
    return Success(a / b)

# 使用示例
result1 = divide_numbers(10, 2)  # Success(5.0)
result2 = divide_numbers(10, 0)  # Failure("除数不能为零")

# 安全地获取结果
if isinstance(result1, Success):
    print(f"计算结果: {result1.unwrap()}")  # 计算结果: 5.0
else:
    print(f"计算错误: {result1.failure()}")

2、Maybe容器处理空值

Maybe容器专门用于处理可能为None的情况，彻底消除程序中的None检查。它有两种状态：Some表示有值，Nothing表示空值。

from returns.maybe import Maybe, Some, Nothing
from typing import Optional

def find_user_by_id(user_id: int) -> Maybe[str]:
    """根据ID查找用户，返回Maybe类型"""
    users = {1: "Alice", 2: "Bob", 3: "Charlie"}
    user = users.get(user_id)
    return Maybe.from_optional(user)

def get_user_email(username: str) -> Maybe[str]:
    """获取用户邮箱"""
    emails = {"Alice": "alice@example.com", "Bob": "bob@example.com"}
    email = emails.get(username)
    return Maybe.from_optional(email)

# 链式操作示例
user_email = (
    find_user_by_id(1)
    .bind(get_user_email)
)

if isinstance(user_email, Some):
    print(f"用户邮箱: {user_email.unwrap()}")  # 用户邮箱: alice@example.com
else:
    print("未找到用户或邮箱信息")

高级功能

1、@safe装饰器和异常处理

@safe装饰器是returns库的强大功能，能够自动捕获函数中的异常并将其转换为Result类型，实现异常安全的函数式编程风格：

from returns.result import Result, Success, Failure, safe
import json

@safe
def parse_json_config(config_str: str) -> dict:
    """安全解析JSON配置"""
    return json.loads(config_str)

@safe
def read_file_content(filename: str) -> str:
    """安全读取文件内容"""
    with open(filename, 'r', encoding='utf-8') as file:
        return file.read()

def load_and_parse_config(filename: str) -> Result[dict, Exception]:
    """链式处理配置文件"""
    return (
        read_file_content(filename)
        .bind(parse_json_config)
        .map(lambda config: config.get('database', {}))
    )

# 使用示例
config_result = load_and_parse_config('config.json')

match config_result:
    case Success(value):
        print(f"数据库配置: {value}")  # 数据库配置: {'host': 'localhost', 'port': 5432, 'username': 'admin', 'password': 'secret'}
    case Failure(error):
        print(f"配置加载失败: {error}")

2、do-notation语法糖

do-notation提供了类似Haskell的语法糖，让复杂的嵌套操作变得更加直观和易读：

from returns.result import Result, Success, Failure
from returns.pipeline import is_successful


class UserService:
    def create_user_account(self, username: str, email: str) -> Result[dict, str]:
        """创建用户账户的复杂流程"""
        result = self._validate_user_data(username, email)
        if not is_successful(result):
            return result

        user_account = self._create_account(result.unwrap())
        if not is_successful(user_account):
            return user_account

        user_profile = self._create_profile(user_account.unwrap())
        if not is_successful(user_profile):
            return user_profile

        return Success({
            'account': user_account.unwrap(),
            'profile': user_profile.unwrap(),
            'status': 'created'
        })

    def _validate_user_data(self, username: str, email: str) -> Result[dict, str]:
        if len(username) < 3:
            return Failure("用户名长度不能少于3个字符")
        if '@' not in email:
            return Failure("邮箱格式不正确")
        return Success({'username': username, 'email': email})

    def _create_account(self, data: dict) -> Result[dict, str]:
        # 模拟账户创建逻辑
        return Success({'id': 12345, 'username': data['username']})

    def _create_profile(self, account: dict) -> Result[dict, str]:
        # 模拟用户资料创建逻辑
        return Success({'user_id': account['id'], 'created_at': '2025-06-28'})

实际应用场景

1、Web API错误处理

在Web应用开发中，returns库可以优雅地处理API请求和响应，确保错误信息的准确传递和处理。

from returns.result import Result, Success, Failure, safe
from returns.pipeline import flow, is_successful
from returns.pointfree import bind
import requests
from typing import Dict, Any


class APIClient:
    def __init__(self, base_url: str):
        self.base_url = base_url

    @safe
    def fetch_user_data(self, user_id: int) -> Dict[str, Any]:
        """安全获取用户数据"""
        response = requests.get(f"{self.base_url}/users/{user_id}")
        response.raise_for_status()
        return response.json()

    def process_user_order(self, user_id: int, order_data: dict) -> Result[dict, str]:
        """处理用户订单的完整流程"""
        return flow(
            Success((user_id, order_data)),  # Start with a Success container
            bind(self._fetch_and_validate),  # Fetch and validate
            bind(self._create_order),  # Create order
            bind(self._send_confirmation_email)  # Send confirmation
        )

    def _fetch_and_validate(self, data: tuple[int, dict]) -> Result[dict, str]:
        """获取用户数据并验证权限"""
        user_id, order_data = data
        user_result = self.fetch_user_data(user_id)

        if not is_successful(user_result):
            return Failure("获取用户数据失败")

        user = user_result.unwrap()
        if not user.get('is_active'):
            return Failure("用户账户未激活")
        if user.get('balance', 0) < order_data.get('total', 0):
            return Failure("账户余额不足")
        return Success({'user': user, 'order': order_data})

    def _create_order(self, data: dict) -> Result[dict, str]:
        """创建订单逻辑"""
        order = {
            'id': 'ORDER-001',
            'user_id': data['user']['id'],
            'total': data['order']['total'],
            'status': 'created'
        }
        return Success({**data, 'order_details': order})

    def _send_confirmation_email(self, data: dict) -> Result[dict, str]:
        """发送确认邮件"""
        # 模拟邮件发送逻辑
        return Success({**data, 'email_sent': True})


# 使用示例
client = APIClient("https://api.example.com")
result = client.process_user_order(123, {'total': 99.99})

match result:
    case Success(value):
        order = result.unwrap()
        print(f"订单创建成功: {order['order_details']['id']}")
    case Failure(value):
        error_message = result.failure()
        print(f"订单处理失败: {error_message}")

2、数据处理管道

在数据处理和ETL场景中，returns库能够构建健壮的数据处理管道，每个步骤都具有明确的错误处理机制。

from returns.result import Result, Success, Failure
from returns.pipeline import flow
from returns.pointfree import bind
import pandas as pd
from typing import Dict, Any, List


class DataProcessor:
    def process_sales_data(self, file_path: str) -> Result[Dict[str, Any], str]:
        """处理销售数据的完整管道"""
        return flow(
            file_path,
            self._load_csv_data,
            bind(self._validate_data_format),
            bind(self._clean_missing_values),
            bind(self._calculate_metrics),
            bind(self._generate_report)
        )

    def _load_csv_data(self, file_path: str) -> Result[pd.DataFrame, str]:
        """加载CSV数据"""
        try:
            df = pd.read_csv(file_path)
            if df.empty:
                return Failure("数据文件为空")
            return Success(df)
        except FileNotFoundError:
            return Failure(f"文件未找到: {file_path}")
        except Exception as e:
            return Failure(f"数据加载失败: {str(e)}")

    def _validate_data_format(self, df: pd.DataFrame) -> Result[pd.DataFrame, str]:
        """验证数据格式"""
        required_columns = ['date', 'product', 'quantity', 'price']
        missing_columns = [col for col in required_columns if col not in df.columns]

        if missing_columns:
            return Failure(f"缺少必要列: {', '.join(missing_columns)}")
        return Success(df)

    def _clean_missing_values(self, df: pd.DataFrame) -> Result[pd.DataFrame, str]:
        """清理缺失值"""
        initial_rows = len(df)
        df_cleaned = df.dropna()
        removed_rows = initial_rows - len(df_cleaned)

        if removed_rows > initial_rows * 0.5:
            return Failure(f"数据质量过低，移除了{removed_rows}行数据")
        return Success(df_cleaned)

    def _calculate_metrics(self, df: pd.DataFrame) -> Result[Dict[str, float], str]:
        """计算业务指标"""
        try:
            df['total_sales'] = df['quantity'] * df['price']
            metrics = {
                'total_revenue': df['total_sales'].sum(),
                'avg_order_value': df['total_sales'].mean(),
                'total_orders': len(df),
                'unique_products': df['product'].nunique()
            }
            return Success(metrics)
        except Exception as e:
            return Failure(f"指标计算失败: {str(e)}")

    def _generate_report(self, metrics: Dict[str, float]) -> Result[Dict[str, Any], str]:
        """生成分析报告"""
        report = {
            'summary': metrics,
            'insights': [],
            'recommendations': []
        }

        # 添加业务洞察
        if metrics['avg_order_value'] > 100:
            report['insights'].append("平均订单价值较高，客户质量良好")
        else:
            report['insights'].append("平均订单价值有提升空间")

        if metrics['unique_products'] < 10:
            report['recommendations'].append("考虑扩大产品线以增加销售机会")
        else:
            report['recommendations'].append("当前产品线多样化程度良好")

        return Success(report)


# 使用示例
processor = DataProcessor()
result = processor.process_sales_data('sales_data.csv')

match result:
    case Success(value):
        report = result.unwrap()
        print(f"处理完成，总收入: {report['summary']['total_revenue']:.2f}")
        print(f"平均订单价值: {report['summary']['avg_order_value']:.2f}")
        for insight in report['insights']:
            print(f"洞察: {insight}")
        for recommendation in report['recommendations']:
            print(f"建议: {recommendation}")
    case Failure(value):
        print(f"数据处理失败: {result.failure()}")

总结

Python returns库为函数式编程和类型安全的错误处理提供了完善的解决方案，通过引入Result、Maybe、IO等容器类型，有效解决了传统Python代码中异常处理复杂、None值检查繁琐等问题。该库的核心优势在于其完整的类型注解支持和与mypy的无缝集成，使开发者能够在享受函数式编程优雅性的同时，保持严格的类型安全保证。通过@safe装饰器、do-notation语法糖等高级功能，returns库大大简化了复杂业务逻辑的编写和维护，提高了代码的可读性和健壮性。