adt.py

# -*- coding:utf-8 -*-
"""
Result code message value = Failure code message | Success value
match x => Failure code, message -> ("oop" code, message) | Success value -> ("ok" value)
"""

from collections import namedtuple
import inspect

class NotComprehensive(Exception):
    pass

class ADTTag(object):
    def __init__(self, tag):
        self.tag = tag
        self.members = {}

    def as_member(self, cls):
        name = cls.__name__
        self.members[name] = cls
        return cls

    def is_comprehensive(self, candidates):
       return (all(m in candidates for m in self.members.keys())
               and len(candidates) == len(self.members))

    def validation(self, cls):
        candidates = [k for k, v in cls.__dict__.items() if is_dispatchmethod(v)]
        if not self.is_comprehensive(candidates):
            raise NotComprehensive("{} != {}".format(candidates, self.members.keys()))
        return cls

    def __call__(self, name, fields):
        return self.as_member(namedtuple(name, fields))

def dispatchmethod(f):
    v = staticmethod(f)
    v._t_dispatch = True
    return v

def is_dispatchmethod(f):
    return hasattr(f, "_t_dispatch")


## factory(experimental)

_i = 0
def gensym():
    global _i
    v = _i
    _i += 1
    return v

def simple_match(tag, **kwargs):
    ## assertion
    for k, fn in kwargs.items():
        try:
            target_type = ResultTag.members[k]
            lhs = tuple(target_type._fields)
            rhs = tuple(inspect.getargspec(fn).args)
            if lhs != rhs:
                raise NotComprehensive("invalida args. {} != {}".format(lhs, rhs))
        except KeyError as e:
            raise NotComprehensive(e)

    attrs = {k:dispatchmethod(fn) for k, fn in kwargs.items()}
    name = "Match_{i}".format(i=gensym())
    def __new__(cls, target):
        tag = target.__class__.__name__
        return getattr(cls, tag)(*target)
    attrs["__new__"] = __new__
    return tag.validation(type(name, (object, ), attrs))


## sample

ResultTag = ADTTag("Result")

Failure = ResultTag("Failure", "code message")
Success = ResultTag("Success", "value")


## define by hand

@ResultTag.validation
class Match0(object):
    @dispatchmethod
    def Failure(code, message):
        return ["oops", code, message]

    @dispatchmethod
    def Success(value):
        return ["ok", value]

    def __new__(cls, target):
        tag = target.__class__.__name__
        return getattr(cls, tag)(*target)


assert ResultTag.is_comprehensive(["Failure", "Success"])
assert Match0(Failure(404, "not found")) == ["oops", 404, "not found"]
assert Match0(Success('{"result": "good"}')) == ["ok", '{"result": "good"}']


## define by factory

Match_0 = simple_match(ResultTag,
                       Success=lambda value: ["ok", value],
                       Failure=lambda code, message: ["oops", code, message]
)

assert Match_0(Failure(404, "not found")) == ["oops", 404, "not found"]
assert Match_0(Success('{"result": "good"}')) == ["ok", '{"result": "good"}']