Rust 言語の Result 型に範を仰いだ、Python 向けの高機能かつ型安全なエラーハンドリングライブラリです。
従来の例外駆動型（Exception-driven）から、Result 駆動型（Result-driven）へのパラダイムシフトを強力に支援します。明示的なエラーハンドリングを導入することで、コードの堅牢性と可読性を飛躍的に向上させることが可能です。
- 厳密な型安全性: ジェネリクスを活用し、成功値とエラー値の双方に対して静的解析（mypy, pyright 等）を適用可能です。
- 鉄道指向プログラミング（ROP）:
mapやand_thenによるメソッドチェーンにより、宣言的なエラーハンドリングを実現します。 - 非同期処理のネイティブサポート:
@async_resultデコレータを通じて、非同期処理をAwaitableResultとして透過的にチェーン可能です。 - メタプログラミングによる統合:
@safeデコレータを用いることで、既存の例外送出型関数を容易に Result 型へ変換できます。 - 高度な結果集約:
combine（早期失敗）およびcombine_all（全エラー集約）により、複数の処理結果を合理的に統合します。 - 軽量設計（ゼロ依存）: 外部ライブラリへの依存はなく、プロジェクトへの導入障壁が極めて低く抑えられています。
pip install flow_res※ Python 3.13 以上が必要です。
関数の戻り値に Result を指定することで、呼び出し側に対してエラー処理の検討を強制（明示）させます。Python 3.10 以降の構造的パターンマッチングを利用することで、エレガントに結果を処理できます。
from flow_res import Result, Ok, Err
def divide(a: int, b: int) -> Result[float, ValueError]:
"""2つの数値の除算を行い、結果を Result 型で返却する"""
if b == 0:
return Err(ValueError("Division by zero"))
return Ok(a / b)
result = divide(10, 2)
match result:
case Ok(value):
print(f"Success: {value}")
case Err(error):
print(f"Failure: {error}")map や and_then を用いることで、命令的な条件分岐を排除し、処理のパイプラインを構築できます。
from flow_res import Result
def validate_positive(x: int) -> Result[int, ValueError]:
if x < 0:
return Err(ValueError("Must be positive"))
return Ok(x)
# 依存関係のある処理の連結
result = (
Ok(5)
.and_then(validate_positive)
.map(lambda x: x * 2)
.map(lambda x: x + 3)
)
print(result.unwrap()) # 13既存の例外を発生させる可能性のある関数を、低コストで Result 駆動型へ移行させます。
from flow_res import safe
@safe
def parse_int(s: str) -> int:
return int(s)
# 例外は送出されず、Err として返却される
result = parse_int("not_a_number")
print(result) # Err(error=ValueError("invalid literal for int() with base 10: 'not_a_number'"))@async_result デコレータを使用することで、非同期関数の実行結果に対しても await 前にメソッドチェーンを適用できます。
import asyncio
from flow_res import Result, async_result
@async_result
async def fetch_user(user_id: int) -> Result[dict, ValueError]:
await asyncio.sleep(0.1)
if user_id < 0:
return Err(ValueError("Invalid user ID"))
return Ok({"id": user_id, "name": f"User{user_id}"})
async def main():
# 処理を連結した後に一括で await
result = await (
fetch_user(1)
.map(lambda u: u["name"])
.map(str.upper)
)
print(result.unwrap()) # USER1
asyncio.run(main())バリデーションなど、複数の検証結果を一括で扱うためのインターフェースを提供します。
combine: 最初に遭遇したErrを返却する（短絡評価・早期失敗）combine_all: すべてのErrを集約して複数の例外を保持するErrを返却する（全件チェック）
from flow_res import Result, combine, combine_all, Ok, Err
results = (
Ok(1),
Err(ValueError("error1")),
Err(RuntimeError("error2")),
)
# 最初のエラー (error1) のみを返す
print(combine(results))
# すべてのエラーを集約して返す
match combine_all(results):
case Err(error):
for e in error.exceptions:
print(f"Error: {e}")- Python バージョン: 3.13 以上
- 型ヒント: 完全対応（Static Type Checking を推奨）
本プロジェクトは MIT License の下に公開されています。
不具合報告や機能拡張の提案は、GitHub Issues にて承っております。