Katharos

Katharos

A functional programming and concurrency library for Python. Chain operations that can fail or return missing data, control when side effects run, and coordinate concurrent workers through typed channels.

Installation

Requires Python 3.13+.

pip install katharos

Or using uv:

uv add katharos

Quick start

Turn exceptions into values and transform successful results:

from katharos.types import Result

@Result.catch(ValueError)
def parse_int(raw: str) -> int:
    return int(raw)

print(parse_int("21").fmap(lambda n: n * 2))
# Success(42)

print(parse_int("oops").fmap(lambda n: n * 2))
# Failure(ValueError("invalid literal for int() with base 10: 'oops'"))

Successful values continue through the pipeline; failures skip subsequent transformations. Result.catch catches the declared exception type and preserves its traceback.

Type hierarchy

Katharos has two independent hierarchies: one for transforming and chaining computations, and one for combining values. Each abstraction adds capabilities to the one above it; concrete types can belong to both.

flowchart TD
    Functor --> Applicative --> Monad
    Semigroup --> Monoid
    Monad -.-> contexts["Maybe, Result, IO, Lazy"]
    Monad -.-> ImmutableList
    Monad -.-> NonEmptyList
    Monoid -.-> ImmutableList
    Semigroup -.-> NonEmptyList
    Monoid -.-> monoids["MonoidMaybe, Sum, Product"]

Solid arrows show inheritance; dotted arrows show implementations. See the type hierarchy reference for methods and laws.

Goals

Katharos exists to make functional-style Python practical, safe, and pleasant to write.

• Errors, absence, and effects as values. Maybe, Result, IO, and Lazy put failure, missing data, and side effects in the type signature, where they compose, instead of hiding them in None and exceptions.
• Law-abiding abstractions. Functor, Applicative, Monad, Semigroup, and Monoid come with their algebraic laws, checked by property-based tests (Hypothesis), so you can rely on them when you refactor.
• Pythonic ergonomics. Operators (|, **, >>, @) and do-notation keep functional code readable to Python developers. Everything is fully type-annotated and checked with pyright.
• Message-passing concurrency. Launch concurrent workers and communicate through typed channels. Channel receives return Result values for data, closure, and timeouts.
• Approachable. Tutorials come first, so you can learn one concept at a time by building something useful.

Non-goals

• Not a port of the full Haskell or Scala typeclass ecosystem. Katharos covers a small, well-tested set of abstractions.
• Not a replacement for asyncio. The concurrency layer is thread-based message passing.

What it looks like

Two everyday problems, each shown without and with Katharos.

Missing values: Maybe

Without Katharos, every lookup needs its own None check:

def find_user(user_id: int) -> str | None:
    return {1: "ada", 2: "grace"}.get(user_id)

def find_discount(user: str) -> float | None:
    return {"ada": 0.15}.get(user)

def lookup_discount(user_id: int) -> float | None:
    user = find_user(user_id)  # str | None
    if user is None:
        return None
    return find_discount(user)

print(lookup_discount(1))  # 0.15
print(lookup_discount(2))  # None: no discount
print(lookup_discount(3))  # None: no user

With Katharos, find_user and find_discount return a Maybe, and do-notation short-circuits on Nothing:

from katharos.types import Maybe
from katharos.syntax_sugar import do, DoBlock

def find_user(user_id: int) -> Maybe[str]:
    return Maybe.from_optional({1: "ada", 2: "grace"}.get(user_id))

def find_discount(user: str) -> Maybe[float]:
    return Maybe.from_optional({"ada": 0.15}.get(user))

@do(Maybe)
def lookup_discount(user_id: int) -> DoBlock[Maybe, float]:
    user: str = yield find_user(user_id)
    discount: float = yield find_discount(user)
    return discount

print(lookup_discount(1))  # Just(0.15)
print(lookup_discount(2))  # Nothing(): no discount
print(lookup_discount(3))  # Nothing(): no user

Failures: Result

Without Katharos, each step can raise, so callers must know which exceptions to catch:

def process(raw: str) -> int:
    n = int(raw)                              # may raise ValueError
    if n <= 0:
        raise ValueError(f"{n} is not positive")
    return n

print(process("42"))  # 42
try:
    process("0")
except ValueError as error:
    print(error)  # 0 is not positive

With Katharos, failure is part of the return type, and steps chain with |:

from katharos.types import Result

@Result.catch(ValueError)
def parse_int(raw: str) -> int:
    return int(raw)

def validate_positive(n: int) -> Result[ValueError, int]:
    if n <= 0:
        return Result.Failure(ValueError(f"{n} is not positive"))
    return Result.Success(n)

def process(raw: str) -> Result[ValueError, int]:
    return parse_int(raw) | validate_positive

print(process("42"))  # Success(42)
print(process("0"))   # Failure(ValueError('0 is not positive'))
print(process("??"))  # Failure(ValueError("invalid literal for int() with base 10: '??'"))

Types at a glance

Types support operators according to their abstractions:

Combining with @

The Semigroup operator combines two values of the same type:

from katharos.types import ImmutableList

ImmutableList([1, 2]) @ ImmutableList([3, 4])  # ImmutableList([1, 2, 3, 4])

Concurrency

Katharos provides Go-style message-passing concurrency: launch workers with csp.go and communicate over typed channels. send() returns Success(None) on completion or Failure(ChannelClosedError) if closure prevents completion. recv() returns a Result containing a value, a closure error, or a timeout error.

Channels

from katharos.concurrency.csp import csp

ch = csp.Channel[int](https://github.com/kamalfarahani/katharos/blob/main/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
ch.send(99)             # Failure(ChannelClosedError('send on closed channel'))

Iterating a channel yields values until it is closed. This is the Fibonacci example from the Go Tour: a producer sends values and closes the channel, and the consumer ranges over it.

from katharos.concurrency.csp import Channel, csp

def fibonacci(n: int, c: Channel[int]) -> None:
    x, y = 0, 1
    for _ in range(n):
        c.send(x).unwrap()
        x, y = y, x + y
    c.close()

c = csp.Channel[int](https://github.com/kamalfarahani/katharos/blob/main/capacity=10)
csp.go(fibonacci, 10, c)

for i in c:  # receives until the channel is closed
    print(i)  # 0 1 1 2 3 5 8 13 21 34

Structured concurrency

Used as a context manager, go becomes a scope that joins everything spawned inside it before the block exits:

from katharos.concurrency.csp import csp

def worker(n: int) -> None: ...

with csp.go:                 # scope waits for all work launched inside
    csp.go(worker, 1)
    csp.go(worker, 2)
# both workers have finished here

Backends

The csp runtime uses standard Python threads by default. Supply a different BaseThreadingBackend to customize how workers run and synchronize.

Documentation

Start here: Getting Started with Katharos.

Full tutorials, how-to guides, API reference, and explanations of the mathematical foundations are at katharos.readthedocs.io.

License

MIT