Organize into a library

.gitignore

@@ -0,0 +1,111 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json

monads.py

@@ -1,141 +0,0 @@
-from typing import Callable, Generic, TypeVar
-
-T = TypeVar("T")
-S = TypeVar("S")
-E = TypeVar("E")
-
-
-class Maybe(Generic[T]):
-    def __init__(self) -> None:
-        raise NotImplementedError
-
-    @classmethod
-    def unit(cls, value: T) -> Maybe[T]:
-        return Just(value)
-
-    def bind(self, function: Callable[[T], Maybe[S]]) -> Maybe[S]:
-        if isinstance(self, Just):
-            return function(self.value)
-        else:
-            new: Maybe[S] = Nothing()
-            return new
-
-    def fmap(self, function: Callable[[T], S]) -> Maybe[S]:
-        if isinstance(self, Just):
-            return Just(function(self.value))
-        else:
-            new: Maybe[S] = Nothing()
-            return new
-
-    __rshift__ = bind
-    __mul__ = __rmul__ = fmap
-
-
-class Just(Maybe[T]):
-    def __init__(self, value: T) -> None:
-        self.value = value
-
-    def __repr__(self) -> str:
-        return f"<Just {self.value}>"
-
-
-class Nothing(Maybe[T]):
-    def __init__(self) -> None:
-        ...
-
-    def __repr__(self) -> str:
-        return "<Nothing>"
-
-
-class Result(Generic[T, E]):
-    def __init__(self) -> None:
-        raise NotImplementedError
-
-    @classmethod
-    def unit(cls, value: T) -> Result[T, E]:
-        return Ok(value)
-
-    def bind(self, function: Callable[[T], Result[S, E]]) -> Result[S, E]:
-        if isinstance(self, Ok):
-            return function(self.value)
-        elif isinstance(self, Err):
-            new: Result[S, E] = Err(self.err)
-            return new
-        else:
-            raise TypeError
-
-    def fmap(self, function: Callable[[T], S]) -> Result[S, E]:
-        if isinstance(self, Ok):
-            return Result.unit(function(self.value))
-        elif isinstance(self, Err):
-            new: Result[S, E] = Err(self.err)
-            return new
-        else:
-            raise TypeError
-
-    __rshift__ = bind
-    __mul__ = __rmul__ = fmap
-
-
-class Ok(Result[T, E]):
-    def __init__(self, value: T) -> None:
-        self.value = value
-
-    def __repr__(self) -> str:
-        return f"<Ok {self.value}>"
-
-
-class Err(Result[T, E]):
-    def __init__(self, err: E) -> None:
-        self.err = err
-
-    def __repr__(self) -> str:
-        return f"<Err {self.err}>"
-
-
-def test_value() -> Result[int, str]:
-    return Ok(5)
-
-
-def test_error() -> Result[int, str]:
-    return Err("oops")
-
-
-def test_map() -> Result[str, str]:
-    five: Ok[int, str] = Ok(5)
-    return five.fmap(str)
-
-
-def test_map_infix() -> Result[str, str]:
-    five: Ok[int, str] = Ok(5)
-    return five * str
-
-
-def test_bind() -> Result[int, str]:
-    five: Ok[int, str] = Ok(5)
-    return five.bind(lambda x: Ok(x + 1))
-
-
-def test_bind_infix() -> Result[int, str]:
-    five: Ok[int, str] = Ok(5)
-    return five >> (lambda x: Ok(x + 1))
-
-
-def test_pipeline() -> Result[int, str]:
-    class Frobnicator(object):
-        @classmethod
-        def create(cls, config: dict) -> Result[Frobnicator, str]:
-            return Ok(cls())
-
-        def dostuff(self) -> Result[list, str]:
-            return Ok(["a", "b", "c", "d"])
-
-    def load_config() -> Result[dict, str]:
-        return Ok({"foo": "bar"})
-
-    return (
-        load_config()
-        >> Frobnicator.create
-        >> Frobnicator.dostuff
-        >> (lambda res: Ok(len(res)))
-    )

monads/__init__.py

@@ -0,0 +1,2 @@
+from .maybe import Maybe, Just, Nothing
+from .result import Result, Ok, Err

monads/functor.py

@@ -0,0 +1,10 @@
+from __future__ import annotations
+from typing import Any, Callable, Generic, TypeVar
+
+T = TypeVar("T")
+S = TypeVar("S")
+
+
+class Functor(Generic[T]):
+    def fmap(self, function: Callable[[T], S]) -> Functor[S]:
+        raise NotImplementedError

monads/maybe.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+from typing import Any, Callable, Generic, Optional, TypeVar
+from .monad import Monad
+
+T = TypeVar("T")
+S = TypeVar("S")
+E = TypeVar("E")
+
+
+class Maybe(Monad[T]):
+    def __init__(self) -> None:
+        raise NotImplementedError
+
+    @classmethod
+    def unit(cls, value: T) -> Maybe[T]:
+        return Just(value)
+
+    def bind(self, function: Callable[[T], Maybe[S]]) -> Maybe[S]:
+        if isinstance(self, Just):
+            return function(self.value)
+        else:
+            new: Maybe[S] = Nothing()
+            return new
+
+    def fmap(self, function: Callable[[T], S]) -> Maybe[S]:
+        if isinstance(self, Just):
+            return Just(function(self.value))
+        else:
+            new: Maybe[S] = Nothing()
+            return new
+
+    def withDefault(self, default: T) -> T:
+        if isinstance(self, Just):
+            return self.value
+        else:
+            return default
+
+    __rshift__ = bind
+    __mul__ = __rmul__ = fmap
+
+
+class Just(Maybe[T]):
+    def __init__(self, value: T) -> None:
+        self.value = value
+
+    def __repr__(self) -> str:
+        return f"<Just {self.value}>"
+
+
+class Nothing(Maybe[T]):
+    def __init__(self) -> None:
+        ...
+
+    def __repr__(self) -> str:
+        return "<Nothing>"
+
+
+def maybe(value: T, predicate: Optional[Callable[[T], bool]] = None) -> Maybe[T]:
+    predicate = predicate or (lambda x: x is not None)
+    if predicate(value):
+        return Just(value)
+    else:
+        return Nothing()

monads/monad.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+from typing import Any, Callable, Generic, TypeVar
+
+from .functor import Functor
+
+T = TypeVar("T")
+S = TypeVar("S")
+
+
+class Monad(Functor[T]):
+    @classmethod
+    def unit(cls, value: T) -> Monad[T]:
+        raise NotImplementedError
+
+    # FIXME: Callable return type set to Any, as the proper value
+    # (Monad[S]) is reported as incompatible with subclass
+    # implementations due to a flaw in mypy:
+    # https://github.com/python/mypy/issues/1317
+    def bind(self, function: Callable[[T], Any]) -> Monad[S]:
+        raise NotImplementedError

monads/result.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+from typing import Any, Callable, Generic, TypeVar
+
+T = TypeVar("T")
+S = TypeVar("S")
+E = TypeVar("E")
+
+
+class Monad(Generic[T]):
+    @classmethod
+    def unit(cls, value: T) -> Monad[T]:
+        raise NotImplementedError
+
+    def bind(self, function: Callable[[T], Any]) -> Monad[S]:
+        raise NotImplementedError
+
+
+class Result(Monad[T], Generic[T, E]):
+    def __init__(self) -> None:
+        raise NotImplementedError
+
+    @classmethod
+    def unit(cls, value: T) -> Result[T, E]:
+        return Ok(value)
+
+    def bind(self, function: Callable[[T], Result[S, E]]) -> Result[S, E]:
+        if isinstance(self, Ok):
+            return function(self.value)
+        elif isinstance(self, Err):
+            new: Result[S, E] = Err(self.err)
+            return new
+        else:
+            raise TypeError
+
+    def fmap(self, function: Callable[[T], S]) -> Result[S, E]:
+        if isinstance(self, Ok):
+            return Result.unit(function(self.value))
+        elif isinstance(self, Err):
+            new: Result[S, E] = Err(self.err)
+            return new
+        else:
+            raise TypeError
+
+    def withDefault(self, default: T) -> T:
+        if isinstance(self, Ok):
+            return self.value
+        else:
+            return default
+
+    __rshift__ = bind
+    __mul__ = __rmul__ = fmap
+
+
+class Ok(Result[T, E]):
+    def __init__(self, value: T) -> None:
+        self.value = value
+
+    def __repr__(self) -> str:
+        return f"<Ok {self.value}>"
+
+
+class Err(Result[T, E]):
+    def __init__(self, err: E) -> None:
+        self.err = err
+
+    def __repr__(self) -> str:
+        return f"<Err {self.err}>"
+
+
+def safe(function: Callable[..., T]) -> Callable[..., Result[T, Exception]]:
+    """Wraps a function that may raise an exception.
+
+    e.g.:
+        @safe
+        def bad() -> int:
+            raise Exception("oops")
+
+    """
+
+    def wrapped(*args, **kwargs) -> Result[T, Exception]:
+        try:
+            return Ok(function(*args, **kwargs))
+        except Exception as e:
+            return Err(e)
+
+    return wrapped

setup.cfg

@@ -0,0 +1,5 @@
+[aliases]
+test=pytest
+
+[tool:pytest]
+addopts = --mypy

setup.py

@@ -0,0 +1,9 @@
+from setuptools import setup  # type: ignore
+
+setup(
+    name="Typesafe Monads",
+    version="0.1dev",
+    packages=["monads"],
+    setup_requires=["pytest-runner"],
+    tests_require=["pytest", "mypy", "pytest-mypy"],
+)

tests/test_maybe.py

@@ -0,0 +1,17 @@
+from monads.maybe import Just, Nothing, maybe
+
+
+def test_maybe_none():
+    assert isinstance(maybe(None), Nothing)
+
+
+def test_maybe_something():
+    assert isinstance(maybe(False), Just)
+
+
+def test_maybe_boolean_false():
+    assert isinstance(maybe(False, predicate=bool), Nothing)
+
+
+def test_maybe_boolean_true():
+    assert isinstance(maybe(True, predicate=bool), Just)

tests/test_result.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+from monads.result import Result, Ok, Err, safe
+
+
+def test_value() -> None:
+    result: Result[int, str] = Ok(5)
+    assert 5 == result.withDefault(0)
+
+
+def test_error() -> None:
+    result: Result[int, str] = Err("oops")
+    assert 0 == result.withDefault(0)
+
+
+def test_fmap() -> None:
+    result: Result[int, str] = Ok(5)
+    mapped: Result[str, str] = result.fmap(str)
+    assert "5" == mapped.withDefault("0")
+
+
+def test_fmap_infix() -> None:
+    result: Result[int, str] = Ok(5)
+    mapped: Result[str, str] = result * str
+    assert "5" == mapped.withDefault("0")
+
+
+def test_bind() -> None:
+    result: Result[int, str] = Ok(5)
+    incremented: Result[int, str] = result.bind(lambda x: Ok(x + 1))
+    assert 6 == incremented.withDefault(0)
+
+
+def test_bind_infix() -> None:
+    result: Result[int, str] = Ok(5)
+    incremented: Result[int, str] = result >> (lambda x: Ok(x + 1))
+    assert 6 == incremented.withDefault(0)
+
+
+def test_pipeline() -> None:
+    class Frobnicator(object):
+        @classmethod
+        def create(cls, config: dict) -> Result[Frobnicator, str]:
+            return Ok(cls())
+
+        def dostuff(self) -> Result[list, str]:
+            return Ok(["a", "b", "c", "d"])
+
+    def load_config() -> Result[dict, str]:
+        return Ok({"foo": "bar"})
+
+    result: Result[int, str] = (
+        load_config()
+        >> Frobnicator.create
+        >> Frobnicator.dostuff
+        >> (lambda res: Ok(len(res)))
+    )
+    assert 4 == result.withDefault(0)