Skip to content

minamorl/darkcore

Folders and files

NameName
Last commit message
Last commit date

Latest commit

Β 

History

43 Commits
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 
Β 

Repository files navigation

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

About

A Haskell-like fp library with the darkest energy

Resources

License

Stars

Watchers

Forks

Packages

No packages published

Languages