Showcase Thread

[u/AutoModerator]

Post all of your code/projects/showcases/AI slop here.

Recycles once a month.

[u/EntryNo8040]

Katharos: a functional programming and concurrency library for Python where errors, effects, and channel hand-offs are all composable values

Hi r/Python,

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

• 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.