darkcore 0.5.1

Practical functional programming primitives for Python: Monads, Transformers, and DSL operators for safe business logic

darkcore

darkcore is a lightweight functional programming toolkit for Python.
It brings Functor / Applicative / Monad abstractions, classic monads like Maybe, Either/Result, Reader, Writer, State,
and an expressive operator DSL (|, >>, @) that makes Python feel almost like Haskell.

---

✨ Features

• Functor / Applicative / Monad base abstractions
• Core monads implemented:
  • Maybe — handle missing values
  • Either / Result — safe error handling
  • Validation — accumulate multiple errors
  • Reader — dependency injection / environment
  • Writer — accumulate logs
  • State — stateful computations
• Monad transformers: MaybeT, ResultT, ReaderT, StateT, WriterT
• Utilities: traverse/sequence, Applicative combinators
• Advanced monads: RWST (Reader-Writer-State)
• Operator overloads for concise DSL-style code:
  • | → fmap (map)
  • >> → bind (flatMap)
  • @ → ap (applicative apply)
• High test coverage, Monad law tests included

---

🚀 Installation

pip install darkcore

(or use Poetry)

---

🧪 Quick Examples

Maybe

from darkcore.maybe import Maybe

m = Maybe(3) | (lambda x: x+1) >> (lambda y: Maybe(y*2))
print(m)  # Just(8)

n = Maybe(None) | (lambda x: x+1)
print(n)  # Nothing

---

Result

from darkcore.result import Ok, Err

def parse_int(s: str):
    try:
        return Ok(int(s))
    except ValueError:
        return Err(f"invalid int: {s}")

res = parse_int("42") >> (lambda x: Ok(x * 2))
print(res)  # Ok(84)

res2 = parse_int("foo") >> (lambda x: Ok(x * 2))
print(res2)  # Err("invalid int: foo")

---

Validation: accumulate errors via Applicative

from darkcore.validation import Success, Failure

def positive(x: int):
    return Failure(["non-positive"]) if x <= 0 else Success(x)

v = Success(lambda a: lambda b: a + b).ap(positive(-1)).ap(positive(0))
print(v)  # Failure(['non-positive', 'non-positive'])

# Result would stop at the first failure

Validation is primarily intended for Applicative composition; bind short-circuits like Result and is not recommended for error accumulation scenarios.

Choosing between Result, Either, and Validation

Type	Error shape	Behavior on bind (>>)	Best use case
Result	Typically string/Exception-like	Short-circuits	IO boundaries, failing effects
Either	Domain-typed error	Short-circuits	Domain errors with rich types
Validation	Accumulates via Applicative	Short-circuits monadically	Form-style multi-error accumulation

Note: Validation accumulates errors in Applicative flows (@ / ap, traverse, sequence_*), but monadically (>>) it short-circuits.

Equality of ReaderT / StateT

These transformers represent computations. Equality is extensional: compare results of run under the same environment/state, not object identity.

---

Reader

from darkcore.reader import Reader

get_user = Reader(lambda env: env["user"])
greet = get_user | (lambda u: f"Hello {u}")

print(greet.run({"user": "Alice"}))  # "Hello Alice"

---

Writer

from darkcore.writer import Writer

# list log by default
w = Writer.pure(3).tell(["start"]) >> (lambda x: Writer(x + 1, ["inc"]))
print(w)  # Writer(4, log=['start', 'inc'])

# for non-``list`` logs, pass ``empty`` and ``combine`` explicitly
# ``empty`` provides the identity element and ``combine`` appends logs
w2 = Writer("hi", empty=str, combine=str.__add__).tell("!")
print(w2)  # Writer('hi', log='!')

# omitting these for a non-``list`` log raises ``TypeError``
try:
    Writer("hi", "!")  # missing empty/combine
except TypeError:
    print("expected TypeError")

---

State

from darkcore.state import State

inc = State(lambda s: (s, s+1))
prog = inc >> (lambda x: State(lambda s: (x+s, s)))

print(prog.run(1))  # (3, 2)

Traverse utilities

from darkcore.traverse import traverse_result
from darkcore.result import Ok, Err

def parse_int(s: str):
    try:
        return Ok(int(s))
    except ValueError:
        return Err(f"bad: {s}")

print(traverse_result(["1", "2"], parse_int))  # Ok([1, 2])
print(traverse_result(["1", "x"], parse_int))  # Err("bad: x")

Result short-circuits on the first Err in traverse_* / sequence_*, whereas Validation accumulates errors under Applicative composition.

RWST

from darkcore.rwst import RWST
from darkcore.result import Ok

combine = lambda a, b: a + b

action = RWST.ask(Ok.pure, combine=combine, empty=list).bind(
    lambda env: RWST.tell([env], Ok.pure, combine=combine, empty=list)
)

print(action(1, 0))  # Ok(((None, 0), [1]))

Operator DSL

from darkcore.maybe import Maybe

mf = Maybe(lambda x: x * 2)
mx = Maybe(4)
print((mf @ mx) | (lambda x: x + 1))  # Just(9)

Pattern Matching

from darkcore.result import Ok, Err
from darkcore.maybe import Maybe
from darkcore.either import Right, Left
from darkcore.writer import Writer

def classify(r):
    match r:
        case Ok(v) if v > 10:
            return ("big", v)
        case Ok(v):
            return ("ok", v)
        case Err(e):
            return ("err", e)

def maybe_demo(m):
    match m:
        case Maybe(value=None):
            return "nothing"
        case Maybe(value=v):
            return v

def either_demo(x):
    match x:
        case Right(v):
            return v
        case Left(e):
            return e

w = Writer(3, ["a"], empty=list, combine=lambda a, b: a + b)
match w:
    case Writer(v, log=ls):
        print(v, ls)

---

📖 Integration Example

from darkcore.reader import Reader
from darkcore.writer import Writer
from darkcore.state import State
from darkcore.result import Ok, Err

# Reader: get user from environment
get_user = Reader(lambda env: env.get("user"))

# Result: validate existence
to_result = lambda user: Err("no user") if user is None else Ok(user)

# Writer: log user
log_user = lambda user: Writer(user, [f"got user={user}"])

# State: update counter
update_state = lambda user: State(lambda s: (f"{user}@{s}", s+1))

env = {"user": "alice"}

user = get_user.run(env)
res = to_result(user) >> (lambda u: Ok(log_user(u)))
writer = res.value
print(writer.log)  # ['got user=alice']

out, s2 = update_state(writer.value).run(42)
print(out, s2)  # alice@42 43

---

Why?

• Safer business code
  • Avoid nested try/except and if None checks
  • Express computations declaratively with monads
• Educational value
  • Learn Haskell/FP concepts hands-on in Python
• Expressive DSL
  • |, >>, @ make pipelines concise and clear

---

Development

git clone https://github.com/minamorl/darkcore
cd darkcore
poetry install
poetry run pytest -v --cov=darkcore

---

License

MIT