Katharos: a functional programming and concurrency library…
0 reactions
Katharos: a functional programming and concurrency library for Python where errors, effects, and channel hand-offs are all composable values
I have been building **Katharos**, a functional programming library for Python that recently grew a message-passing concurrency layer. I wanted to share it and get some feedback.
The whole library is built around one idea: model errors, effects, and concurrent communication as **composable, type-safe values** rather than as control flow that jumps around your program. The interesting part (to me, at least) is that the concurrency layer follows the exact same idea, so receiving from a channel gives you a `Result`. "The channel is closed" becomes a value you handle, not an exception you remember to catch.
## The functional core
Optional values without scattered `None` checks, using do-notation that short-circuits on `Nothing`:
```python
from katharos.types import Maybe
from katharos.syntax_sugar import do, DoBlock
def lookup_discount(user_id: int) -> DoBlockMaybe, float]:
user = yield find_user(user_id)
account = yield find_account(user)
return account.discount # Just(0.15) or Nothing()
```
Errors as values, chained with `|`, so a failure short-circuits the rest automatically:
```python
from katharos.types import Result
def process(raw: str) -> Result[Exception, int]:
return parse_int(raw) | validate_positive
```
And `Result.catch` turns a function that raises into one that returns a `Result`, while keeping the original traceback so you can still find the line that failed:
```python
from katharos.types import Result
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: '??'"))
```
There is also `ImmutableList`, `NonEmptyList`, `IO`, `Lazy`, numeric monoids, and the usual algebraic abstractions (`Functor`, `Applicative`, `Monad`, `Semigroup`, `Monoid`) if you want to build your own types.
## The new part: CSP concurrency
This is what I have been working on lately. Katharos now has Go-style CSP (Communicating Sequential Processes): launch work concurrently with `go`, talk over typed channels, and receive values as a `Result`.
```python
from katharos.concurrency.csp import csp
ch = csp.Channel[int capacity=1
csp.go(ch.send, 42) # run work concurrently, like Go's `go f(x)`
ch.recv() # Success(42)
ch.close()
ch.recv() # Failure(ChannelClosedError(...)): closure is a value, not a raise
```
Used as a context manager, `go` becomes a structured-concurrency scope that joins everything spawned inside it before the block exits, so concurrent work cannot leak out of the block:
```python
with csp.go: # scope waits for all work launched inside
csp.go(worker, 1)
csp.go(worker, 2)
# both workers have finished here
```
There is also a `select` for waiting on whichever of several channels is ready first, with non-blocking polls and timeouts:
```python
from katharos.concurrency.csp import csp, recv, select
choice = select(recv(results), recv(cancel), timeout=1.0)
if choice.is_timeout:
...
else:
print(choice.index, choice.value.unwrap())
```
The concurrency model sits on a swappable backend (standard threads by default), so the same code could run on a green-thread backend later. An actor model is planned next, built on the same backend abstraction and the same `Result`-valued style.
## Why I think the "channel returns a Result" thing is nice
In most channel APIs, a closed channel or a timeout shows up as a sentinel, a second return value, or an exception. In Katharos it is just a typed value: `Success(v)`, `Failure(ChannelClosedError)`, or `Failure(ChannelTimeoutError)`. You pattern-match it the same way you handle any other `Result`, and the type tells you it can happen. The error-handling discipline you use in the rest of your code carries straight over to concurrency.
## Links
- PyPI: `pip install katharos`
- Docs (tutorials, how-to guides, API reference, and explanations): https://katharos.readthedocs.io/en/latest/
- Source: https://github.com/kamalfarahani/katharos
- MIT licensed, Python 3.13+
I would love feedback on the API, the concurrency design, or whether the `Result`-everywhere approach feels natural or noisy to you in practice. Thanks for reading.
#technology
I have been building **Katharos**, a functional programming library for Python that recently grew a message-passing concurrency layer. I wanted to share it and get some feedback.
The whole library is built around one idea: model errors, effects, and concurrent communication as **composable, type-safe values** rather than as control flow that jumps around your program. The interesting part (to me, at least) is that the concurrency layer follows the exact same idea, so receiving from a channel gives you a `Result`. "The channel is closed" becomes a value you handle, not an exception you remember to catch.
## The functional core
Optional values without scattered `None` checks, using do-notation that short-circuits on `Nothing`:
```python
from katharos.types import Maybe
from katharos.syntax_sugar import do, DoBlock
def lookup_discount(user_id: int) -> DoBlockMaybe, float]:
user = yield find_user(user_id)
account = yield find_account(user)
return account.discount # Just(0.15) or Nothing()
```
Errors as values, chained with `|`, so a failure short-circuits the rest automatically:
```python
from katharos.types import Result
def process(raw: str) -> Result[Exception, int]:
return parse_int(raw) | validate_positive
```
And `Result.catch` turns a function that raises into one that returns a `Result`, while keeping the original traceback so you can still find the line that failed:
```python
from katharos.types import Result
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: '??'"))
```
There is also `ImmutableList`, `NonEmptyList`, `IO`, `Lazy`, numeric monoids, and the usual algebraic abstractions (`Functor`, `Applicative`, `Monad`, `Semigroup`, `Monoid`) if you want to build your own types.
## The new part: CSP concurrency
This is what I have been working on lately. Katharos now has Go-style CSP (Communicating Sequential Processes): launch work concurrently with `go`, talk over typed channels, and receive values as a `Result`.
```python
from katharos.concurrency.csp import csp
ch = csp.Channel[int capacity=1
csp.go(ch.send, 42) # run work concurrently, like Go's `go f(x)`
ch.recv() # Success(42)
ch.close()
ch.recv() # Failure(ChannelClosedError(...)): closure is a value, not a raise
```
Used as a context manager, `go` becomes a structured-concurrency scope that joins everything spawned inside it before the block exits, so concurrent work cannot leak out of the block:
```python
with csp.go: # scope waits for all work launched inside
csp.go(worker, 1)
csp.go(worker, 2)
# both workers have finished here
```
There is also a `select` for waiting on whichever of several channels is ready first, with non-blocking polls and timeouts:
```python
from katharos.concurrency.csp import csp, recv, select
choice = select(recv(results), recv(cancel), timeout=1.0)
if choice.is_timeout:
...
else:
print(choice.index, choice.value.unwrap())
```
The concurrency model sits on a swappable backend (standard threads by default), so the same code could run on a green-thread backend later. An actor model is planned next, built on the same backend abstraction and the same `Result`-valued style.
## Why I think the "channel returns a Result" thing is nice
In most channel APIs, a closed channel or a timeout shows up as a sentinel, a second return value, or an exception. In Katharos it is just a typed value: `Success(v)`, `Failure(ChannelClosedError)`, or `Failure(ChannelTimeoutError)`. You pattern-match it the same way you handle any other `Result`, and the type tells you it can happen. The error-handling discipline you use in the rest of your code carries straight over to concurrency.
## Links
- PyPI: `pip install katharos`
- Docs (tutorials, how-to guides, API reference, and explanations): https://katharos.readthedocs.io/en/latest/
- Source: https://github.com/kamalfarahani/katharos
- MIT licensed, Python 3.13+
I would love feedback on the API, the concurrency design, or whether the `Result`-everywhere approach feels natural or noisy to you in practice. Thanks for reading.
#technology
| 3,000 mlx | |
| 4 | |
| 0 |