プログラミング
Show HN: Katharos Python向け関数型プログラミングとCSPスタイルの並行処理
Show HN: Katharos Functional programming and CSP-style concurrency for Python (github.com)
要約
Katharosは、Python向けの関数型プログラミングと並行処理ライブラリです。代数的抽象化と具体的な型を、メッセージパッシング型の並行処理と組み合わせています。エラーや副作用、並行処理の通信を合成可能で型安全な値としてモデル化することで、例外処理やNoneチェックの代わりに、よりクリーンで宣言的なコードを実現します。GoスタイルのCSP並行処理もサポートしており、チャネルのクローズやタイムアウトも例外ではなく値として扱います。
全文翻訳
Katharos Python向けの関数型プログラミングと並行処理ライブラリです。
Functor, Applicative, Monad, Semigroup, Monoidのような代数的抽象化と、Maybe, Result, ImmutableList, IOのような具体的な型を、同じ関数型コアに基づいたメッセージパッシング型の並行処理と組み合わせています。
この二つの部分は一つのアイデアを共有しています。それは、エラー、副作用、並行処理の通信を合成可能で型安全な値としてモデル化することです。
並行処理での値の受け渡しはResultを返します。そのため、「チャネルが閉じられた」ということは、キャッチする例外ではなく、あなたが処理すべきものとなります。
インストール
pip install katharos
またはuvを使用する
```bash
uv add katharos
```
どのようなものか
以前:
散在するNoneチェックと例外処理:
```python
user = find_user(user_id)
if user is None:
return None
account = find_account(user)
if account is None:
return None
return account.discount
```
以後:
Nothingでクリーンにショートサーキットするdo-notation:
```python
from katharos.types import Maybe
from katharos.syntax_sugar import do, DoBlock
@do(Maybe)
def lookup_discount(user_id: int) -> DoBlock[Maybe, float]:
user = yield find_user(user_id)
account = yield find_account(user)
return account.discount # Just(0.15) または Nothing()
```
以前:
エラーを伝播させるためのネストされたtry/except:
```python
def process(raw: str) -> int:
try:
n = parse_int(raw)
except ValueError as e:
raise RuntimeError("bad input") from e
try:
return validate_positive(n)
except ValueError as e:
raise RuntimeError("bad value") from e
```
以後:
エラーを値として扱い、|で連鎖させる:
```python
from katharos.types import Result
def process(raw: str) -> Result[Exception, int]:
return parse_int(raw) | validate_positive # Failureは自動的にショートサーキットする
```
その他の例
Noneチェックなしでオプション値を処理する:
```python
from katharos.types import Maybe
result = Maybe[int].Just(5) | (lambda x: Maybe[int].Just(x * 2)) # Just(10)
nothing = Maybe[int].Nothing() | (lambda x: Maybe[int].Just(x * 2)) # Nothing()
```
例外ではなく値としてエラーをモデル化する:
```python
from katharos.types import Result
def parse_int(s: str) -> Result[ValueError, int]:
try:
return Result.Success(int(s))
except ValueError as e:
return Result.Failure(e)
parse_int("42").fmap(lambda n: n * 2) # Success(84)
parse_int("??").fmap(lambda n: n * 2) # Failure(...)
```
Result.catchでボイラープレートをスキップする:
Result.catchは、例外を発生させる関数を、手動のtry/exceptなしでResultを返す関数に変換します。
宣言された例外タイプのみがFailureとなり、キャッチされた例外はそのトレースバックを保持するため、失敗した行を依然として見つけることができます。
```python
import traceback
from katharos.types import Result
@Result.catch(ValueError)
def parse_int(s: str) -> int:
return int(s)
parse_int("42") # Success(42)
parse_int("??") # Failure(ValueError("invalid literal for int() with base 10: '??'"))
failure = parse_int("??")
if failure.is_failure():
traceback.print_exception(failure.error) # 完全なトレースバック、失敗した行を指す
```
Semigroup演算子で値を結合する:
```python
from katharos.types import ImmutableList
ImmutableList([1, 2]) @ ImmutableList([3, 4]) # ImmutableList([1, 2, 3, 4])
```
Do-notation
do-notationは、Maybe, Result, IO, ImmutableList、およびカスタムモナドなど、あらゆるモナドで機能します。
各yieldはモナド値をアンラップします:
```python
from katharos.syntax_sugar import do, DoBlock
from katharos.types import Result
def parse_positive(x: int) -> Result[ValueError, int]:
return Result.Success(x) if x > 0 else Result.Failure(ValueError(f"{x} is not positive"))
# クリーンで命令型のモナドコード
@do(Result)
def do_block() -> DoBlock[Result, int]:
x: int = yield parse_positive(5)
y: int = yield parse_positive(3)
return x + y
print(do_block()) # Success(8)
```
並行処理
Katharosは、同じ関数型コア上に構築されたメッセージパッシング型の並行処理を提供し、複数の並行処理モデルの余地があります。
最初に使用可能なモデルはGoスタイルのCSPです。go（Goの`go f(x)`のようなもの）で並行処理で作業を起動し、型付きチャネルを介して通信し、そして（決定的に重要ですが）値を受信するときはResultとして受信します。そのため、閉じられた、またはタイムアウトしたチャネルは、例外ではなくパターンマッチングで処理する値となります。
```python
from katharos.concurrency.csp import csp
ch = csp.Channel[int](capacity=1)
csp.go(ch.send, 42) # Goの`go f(x)`のように、作業を並行して実行する
ch.recv() # Success(42)
ch.close()
ch.recv() # Failure(ChannelClosedError(...)): クローズは例外ではなく値です
```
コンテキストマネージャーとして使用すると、`go`は構造化された並行処理スコープとなり、ブロックが終了する前にその中で起動されたすべての処理を結合します:
```python
from katharos.concurrency.csp import csp
with csp.go: # スコープは内部で起動されたすべての作業が完了するのを待つ
csp.go(worker, 1)
csp.go(worker, 2)
# ここで両方のワーカーが完了している
```
すべての並行処理モデルは、スワップ可能な`BaseThreadingBackend`（デフォルトでは標準スレッド）にバインドされており、`csp`ランタイムはそれを自動的に供給するため、単一の場所で作業を異なるバックエンドにリターゲットできます。
追加のモデル（アクターモデルなど）は、同じバックエンド抽象化と、同じResult値を持つ合成可能なスタイルに基づいて計画されています。
ドキュメント
完全なチュートリアル、ハウツーガイド、APIリファレンス、および数学的基礎の説明は katharos.readthedocs.io で入手できます。
ライセンス
MIT