Katharos: Functional Programming and CSP-style Concurrency for Python

Katharos is a functional programming and concurrency library for Python 3.13+ that replaces hidden control flow—such as exceptions and None checks—with explicit, composable, and type-safe values. By pairing algebraic abstractions with a message-passing concurrency model, Katharos allows developers to treat errors, optionality, and concurrent communication as first-class values.

Algebraic Abstractions and Type-Safe Values

Katharos implements core functional programming abstractions to eliminate boilerplate and reduce runtime errors associated with null values and exceptions.

Eliminating None Checks with Maybe

Instead of scattered if x is None checks, Katharos uses the Maybe type. This allows for clean short-circuiting of logic when a value is absent.

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  # Returns Just(0.15) or Nothing()

Error Handling as Values with Result

Katharos treats errors as values rather than exceptions. The Result type allows errors to be chained using the pipe operator (|), where a Failure automatically short-circuits the rest of the chain.

To reduce manual try/except boilerplate, the @Result.catch decorator converts functions that raise specific exceptions into functions that return a Result type while preserving the original traceback for debugging.

from katharos.types import Result

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

# Returns Success(42) or Failure(ValueError(...))
result = parse_int("42")

Additional Functional Types

  • ImmutableList: Provides immutable sequences that can be combined using the Semigroup operator (@).
  • Algebraic Core: The library provides formal implementations of Functor, Applicative, Monad, Semigroup, and Monoid to ensure consistent composition across different types.

CSP-Style Concurrency

Katharos implements a Go-style Communicating Sequential Processes (CSP) model for concurrency, ensuring that communication between concurrent tasks is as type-safe as the functional data types.

Message Passing and Channels

Work is launched concurrently using csp.go, and communication occurs via typed Channels. Crucially, receiving a value from a channel returns a Result. This means a closed or timed-out channel is a value to be pattern-matched rather than an exception to be caught.

from katharos.concurrency.csp import csp

ch = csp.Channel[int](capacity=1)

# Run work concurrently
csp.go(ch.send, 42)

# Receive as a Result value
ch.recv()  # Success(42)

ch.close()
ch.recv()  # Failure(ChannelClosedError(...))

Structured Concurrency

When used as a context manager, csp.go creates a structured concurrency scope. The program will join all work spawned within that block before exiting the scope, preventing leaked goroutine-like tasks.

with csp.go:
    csp.go(worker, 1)
    csp.go(worker, 2)
# Both workers are guaranteed to have finished here

Pluggable Backends

All concurrency models in Katharos are bound to a BaseThreadingBackend. While standard threads are the default, this abstraction allows the runtime to be retargeted to different backends in a single location. Future updates are planned to include an actor model built on this same backend abstraction.

Implementation Details and Community Feedback

Do-Notation Mechanism

Katharos provides do-notation via Python generators to allow imperative-style monadic code. However, community feedback has noted that because Python lacks call/cc (call-with-current-continuation), the implementation for certain monads (like the list monad) may involve rerunning the generator, which may limit its use to simpler use cases or "toy" examples.

"It's probably worth being up front about what your do syntax actually does. A generator is not as general as "do", Python doesn't have call/cc or anything like that."

Requirements

Katharos is designed for Python 3.13+.

Sources