New: Introducing PHP and Composer Support.Read the Announcement
Socket
Book a DemoInstallSign in
Socket

darkcore

Package Overview
Dependencies
Maintainers
1
Versions
7
Alerts
File Explorer

Advanced tools

Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

darkcore

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

pipPyPI
Version
0.5.1
Maintainers
1

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

TypeError shapeBehavior on bind (>>)Best use case
ResultTypically string/Exception-likeShort-circuitsIO boundaries, failing effects
EitherDomain-typed errorShort-circuitsDomain errors with rich types
ValidationAccumulates via ApplicativeShort-circuits monadicallyForm-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

Keywords

monad

FAQs

Did you know?

Socket

Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.

Install

Related posts