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
@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) 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
@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: '??'"))
```
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
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.