<jupyter_start><jupyter_text>Ideas- more generalizable version of _apply so that any op can be applied to obj, not just mapping some func on self.vals    - inplace param also needs to generalize- figure out how I want decorator to work (or no decorator?). Don't love current implementation of having to return a lambda    - maybe chainable should be a method of Chained?- still have to think about how this could be used: mixin class? Would a metaclass be useful here? Have to think.<jupyter_code>from functools import wraps, partial
def chain(func):
    @wraps(func)
    def wrapped(instance, *args, **kwargs):
        return instance._apply(func(*args, **kwargs))
    return wrapped
class Chainable:
   
    def __init__(self, vals):
        self.vals = vals
        self.ops = []
    
    def _apply(self, func, mode='map'):
        op = map if mode == 'map' else filter
        self.ops.append((op, func))
        return self
        
    def exec(self, inplace=False):
        new = self
        for op, func in self.ops:
            new = Chainable(list(op(func, new.vals)))
        self.ops.clear()
        if inplace: 
            self.__dict__ = new.__dict__
        else: 
            return new
        
    def double(self):
        return self._apply(lambda x: x*2)
        
    def add(self, n):
        return self._apply(lambda x: x+n)
        
    def stringify(self):
        return self._apply(str)
        
    def gt(self, n):
        return self._apply(lambda x: x > n, 'filter')
        
    def even_only(self):
        return self._apply(lambda x: x % 2 == 0, 'filter')
    
    @chain
    def subtract(n):
        return lambda x: x-n
        
    def __repr__(self):
        return f'Chainable({repr(self.vals)})'
c = Chainable([1, 3, 2, 44, -5, -7, 16, 6.6, 9, -3, 0.5, 1])
c
c.double().exec()
c.double().even_only().gt(-5).stringify().exec()
c.subtract(4).double().exec()
c
c.add(4).subtract(2).even_only().gt(-1).exec(inplace=True)
c<jupyter_output><empty_output><jupyter_text>Working on more generalized versionCurrent implementation: user defines a staticmethod for each chainable op, then an instance method that calls that staticmethod. This works but is annoying to use.<jupyter_code>from copy import copy, deepcopy
from functools import update_wrapper, wraps
import inspect
class Chainable():
   
    def __init__(self, vals, num):
        self.vals = vals
        self.been_called = False
        self.num = num
        self.ops = []
    
    def _apply(self, func):
        self.ops.append(func)
        return self
        
    def exec(self, inplace=False):
        new = deepcopy(self)
        for func in self.ops:
            new = func(copy(new))

        # Clear ops list now that chain is complete.
        new.ops.clear()
        if inplace: 
            self.__dict__ = new.__dict__
        else: 
            self.ops.clear()
            return new
        
    @staticmethod
    def _double(instance):
        instance.num *= 2
        return instance
    
    def double(self):
        return self._apply(self._double)
        
    @staticmethod
    def _add(instance, n):
        instance.num += n
        return instance
    
    def add(self, n):
        return self._apply(partial(self._add, n=n))

    @staticmethod    
    def _append(instance, n):
        instance.vals.append(n)
        return instance

    def append(self, n):
        return self._apply(partial(self._append, n=n))
        
    @staticmethod
    def _call(instance):
        instance.been_called = True
        return instance
    
    def call(self):
        return self._apply(self._call)
    
    def __repr__(self):
        return f'Chainable({repr(self.vals)}, {self.been_called}, {self.num}, {self.ops})'
c1 = Chainable([1, 3], 5)
c1
c2 = copy(c1)
c2
id(c1.vals), id(c2.vals)
c2.vals.append(3)
c1, c2
c1.double().exec(True)
c1, c2
c2.append(-1).exec(True)
c1, c2<jupyter_output><empty_output><jupyter_text>Testing basic func<jupyter_code>vals = [5, 3, 1, 2, 4, 6]
n = 44
length = len(vals)
c = Chainable(vals, n)
c
c.add(3).double().exec()
print(c)

assert c.vals == vals and not c.been_called and c.num == n and not c.ops
c.add(3).double().exec(inplace=True)
print(c)
assert c.vals == vals and not c.been_called and c.num == (n+3)*2 and not c.ops<jupyter_output>Chainable([5, 3, 1, 2, 4, 6], False, 94, [])<jupyter_text>Testing effect on mutable attrs<jupyter_code>c.append(99).call().exec(False)
# Bool attr is unchanged (as desired), but list is changed.
print(c)
assert c.vals == vals and not c.been_called and c.num == (n+3)*2 and not c.ops
c.append(99).call().exec(True)
# Bool attr and list attr both changed.
print(c)
assert c.vals == vals+[99] and c.been_called and c.num == (n+3)*2 and not c.ops
assert len(vals) == length<jupyter_output><empty_output><jupyter_text>Testing meta magicIdea: can we make it simpler or more intuitive to define chainable methods? Status: Tentatively working, needs testing though.<jupyter_code>import types

from htools import hdir, debug
# class ChainMethod:
    
#     def __init__(self, func):
#         wraps(func)(self)
        
#     def __call__(self, *args, **kwargs):
#         print('call', args, kwargs)
#         return self.__wrapped__(*args, **kwargs)
    
#     def __get__(self, instance, cls):
#         if instance is None:
#             return self
#         return types.MethodType(self, instance)

    
def chain(func):
    func._is_chainable = True
    
    @wraps(func)
    def wrapped(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapped


# def ChainMethod(func):
#     # This version does attach _is_trainable attr but v.__name__
#     # is now wrapper in metaclass
    
#     @wraps(func)
#     def wrapper(*args, **kwargs):
#         print('CALL DECORATOR: args', args, '\nkwargs', kwargs)
#         return func(*args, **kwargs)
#     static = staticmethod(wrapper)
#     static._is_chainable = True
#     return static
class ChainMeta(type):
    
    def __new__(cls, name, bases, methods):
        new_methods = {}
        
        for k, v in methods.items():
            try: 
                func = v.__get__(1)
                assert func._is_chainable
            except: 
                continue    
            public_name = k.lstrip('_')
            
            # Capture args and kwargs passed to staticmethod (except for instance).
            sig = inspect.signature(func)
            sig = sig.replace(parameters=list(sig.parameters.values())[1:])
            
            # Must use default args, otherwise func will always point to last method.
            def make_public_method(func=func, private_name=k, 
                                   public_name=public_name, sig=sig):
                def public(inst, *args, **kwargs):
                    bound = sig.bind(*args, **kwargs).arguments
                    new_method = partial(getattr(inst, private_name), **bound)
                    inst.ops.append(new_method)
                    return inst
                public.__name__ = public_name
                return public
            
            new_methods[public_name] = make_public_method()
                
        return type.__new__(cls, name, bases, {**methods, **new_methods})
class Chainable(metaclass=ChainMeta):
   
    def __init__(self, vals, num):
        self.vals = vals
        self.been_called = False
        self.num = num
        self.ops = []
    
    def _apply(self, func):
        self.ops.append(func)
        return self
        
    def exec(self, inplace=False):
        new = deepcopy(self)
        for func in self.ops:
            new = func(copy(new))
        # Clear ops list now that chain is complete.
        new.ops.clear()
        if inplace: 
            self.__dict__ = new.__dict__
        else: 
            self.ops.clear()
            return new
    
    @staticmethod
    @chain
    def _call(instance):
        instance.been_called = True
        return instance
    
    @staticmethod
    @chain
    def _incr(instance):
        instance.num += 1
        return instance
    
    @staticmethod
    @chain
    def _append(instance, n):
        instance.vals.append(n)
        return instance
    
    def __repr__(self):
        return f'Chainable({repr(self.vals)}, {self.been_called}, {self.num}, {self.ops})'
c1 = Chainable([1, 3], 5)
c1
assert not c1.been_called
_ = c1.call().exec(True)
assert _ is None
assert c1.been_called
assert c1.num == 5
c1.incr().exec()
assert c1.num == 5
_ = c1.incr().exec(True)
assert _ is None
assert c1.num == 6
c1 = Chainable([2, 4, 6], 111)
c1
c2 = c1.incr().call().exec()
assert c2.been_called
assert c2.num == 112

assert not c1.been_called
assert c1.num == 111
_ = c1.incr().call().append(n=99).exec(True)
assert _ is None
assert c1.vals == [2, 4, 6, 99]
assert c1.been_called
assert c1.num == 112<jupyter_output><empty_output><jupyter_text>Issue: only kwargs work atm, not args. Must map args to correct params in partial in metaclass<jupyter_code>c2 = c1.append(4).exec()
assert c2.vals[-1] == 4
assert c1.vals[-1] == 99
c1
c1.append(333).incr().call().exec()
c1
c1.append(333).incr().call().exec(True)
c1<jupyter_output><empty_output><jupyter_text>Testing finished lazy chainableNote: Library version has a few changes to formatting (line lengths), docstrings, and removed \_apply() method in an effort to enforce 1 correct way of making chainable methods. chain decorator is also renamed to lazychain. IDEA: check if staticmethod is actually necessary now. May have been left over from a previous approach.<jupyter_code>def chain(func):
    """Decorator to register a method as chainable within a
    LazyChainable class.
    """
    func._is_chainable = True
    
    @wraps(func)
    def wrapped(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapped
class LazyChainMeta(type):
    """Metaclass to create LazyChainable objects."""
    
    def __new__(cls, name, bases, methods):
        new_methods = {}
        
        # Find chainable staticmethods and create public versions.
        for k, v in methods.items():
            try: 
                func = v.__get__(1)
                assert func._is_chainable
            except: 
                continue    
            public_name = k.lstrip('_')
            
            # Capture args and kwargs passed to staticmethod (except for instance).
            sig = inspect.signature(func)
            sig = sig.replace(parameters=list(sig.parameters.values())[1:])
            
            # Must use default args, otherwise func will always point to last method.
            def make_public_method(func=func, private_name=k, 
                                   public_name=public_name, sig=sig):
                def public(inst, *args, **kwargs):
                    bound = sig.bind(*args, **kwargs).arguments
                    new_method = partial(getattr(inst, private_name), **bound)
                    inst.ops.append(new_method)
                    return inst
                public.__name__ = public_name
                return public
            
            new_methods[public_name] = make_public_method()
                
        return type.__new__(cls, name, bases, {**methods, **new_methods})
class LazyChainable(metaclass=LazyChainMeta):
    """Base class that allows children to lazily chain methods, 
    similar to a Spark RDD.
    
    Chainable methods must be decorated with @staticmethod
    and @chain and be named with a leading underscore. A public
    method without the leading underscore will be created, so don't
    overwrite this with another method. Chainable methods
    accept an instance of the same class as the first argument, 
    process the instance in some way, then return it. A chain of 
    commands will be stored until the exec() method is called.
    It can operate either in place or not.
    
    Examples
    --------
    class Sequence(LazyChainable):
    
        def __init__(self, numbers, counter, new=True):
            super().__init__()
            self.numbers = numbers
            self.counter = counter
            self.new = new

        @staticmethod
        @chain
        def _sub(instance, n):
            instance.counter -= n
            return instance

        @staticmethod
        @chain
        def _gt(instance, n=0):
            instance.numbers = list(filter(lambda x: x > n, instance.numbers))
            return instance

        @staticmethod
        @chain
        def _call(instance):
            instance.new = False
            return instance

        def __repr__(self):
            pre, suf = super().__repr__().split('(')
            argstrs = (f'{k}={repr(v)}' for k, v in vars(self).items())
            return f'{pre}({", ".join(argstrs)}, {suf}'
    
    
    >>> seq = Sequence([3, -1, 5], 0)
    >>> output = seq.sub(n=3).gt(0).call().exec()
    >>> output
    
    Sequence(ops=[], numbers=[3, 5], counter=-3, new=False)
    
    >>> seq   # Unchanged because exec was not in place.
    
    Sequence(ops=[], numbers=[3, -1, 5], counter=0, new=True)
    
    
    >>> output = seq.sub(n=3).gt(-1).call().exec(inplace=True)
    >>> output   # None because exec was in place.
    >>> seq   # Changed
    
    Sequence(ops=[], numbers=[3, -1, 5], counter=-3, new=False)
    """
   
    def __init__(self):
        self.ops = []
    
    def _apply(self, func):
        self.ops.append(func)
        return self
        
    def exec(self, inplace=False):
        new = deepcopy(self)
        for func in self.ops:
            new = func(copy(new))
        # Clear ops list now that chain is complete.
        new.ops.clear()
        if inplace: 
            self.__dict__ = new.__dict__
        else: 
            self.ops.clear()
            return new
    
    def __repr__(self):
        argstrs = (f'{k}={repr(v)}' for k, v in vars(self).items())
        return f'{type(self).__name__}({", ".join(argstrs)})'
class Sequence(LazyChainable):
    
    def __init__(self, numbers, counter, new=True):
        super().__init__()
        self.numbers = numbers
        self.counter = counter
        self.new = new
        
    @staticmethod
    @chain
    def _sub(instance, n):
        instance.counter -= n
        return instance
    
    @staticmethod
    @chain
    def _product(instance):
        prod = 1
        for num in instance.numbers:
            prod *= num
        instance.prod = prod
        return instance
    
    @staticmethod
    @chain
    def _gt(instance, n=0):
        instance.numbers = list(filter(lambda x: x > n, instance.numbers))
        return instance
    
    @staticmethod
    @chain
    def _call(instance):
        instance.new = False
        return instance
    
    def __repr__(self):
        return super().__repr__()
seq = Sequence([1, 11, 99, 4, -3, -0.5, 1.5, -22.2], 4, new=True)
seq
seq.product().exec()
seq.gt(4).sub(3).product().exec(True)
seq
seq.ops
seq.sub(n=4).call().product().exec()
isinstance(seq, LazyChainable)
type(LazyChainable)<jupyter_output><empty_output><jupyter_text>Try another approach using \_\_getattr\_\_ instead of metaclass<jupyter_code>def ChainMethod(func):
    func._is_chainable = True
    
    @wraps(func)
    def wrapped(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapped
class Chainable:
   
    def __init__(self, vals, num):
        self.vals = vals
        self.been_called = False
        self.num = num
        self.ops = []
        
    def __getattr__(self, name, *args, **kwargs):
        print('args, kwargs', args, kwargs)
        private_name = '_' + name
        method = super().__getattribute__(private_name)
        if method._is_chainable:
            self._apply(method)
        return method
    
    def _apply(self, func):
        self.ops.append(func)
        return self
        
    def exec(self, inplace=False):
        new = deepcopy(self)
        for func in self.ops:
            new = func(copy(new))
        # Clear ops list now that chain is complete.
        new.ops.clear()
        if inplace: 
            self.__dict__ = new.__dict__
        else: 
            self.ops.clear()
            return new
    
    @staticmethod
    @ChainMethod
    def _call(instance):
        instance.been_called = True
        return instance
    
    @staticmethod
    @ChainMethod
    def _incr(instance):
        instance.num += 1
        return instance
    
    @staticmethod
    @ChainMethod
    def _append(instance, n):
        instance.vals.append(n)
        return instance
    
    def __repr__(self):
        return f'Chainable({repr(self.vals)}, {self.been_called}, {self.num}, {self.ops})'
c = Chainable([2, 4, 0], 100)
c
c.incr()
c.append(n=3)<jupyter_output>args, kwargs () {}
{'fraction_non_alphanumeric': 0.11426692178904568, 'fraction_numerical': 0.012018655823965559, 'mean_word_length': 2.191793893129771, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 55, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '487409', 'license': 'MIT', 'n_tokens_mistral': 5269, 'n_tokens_neox': 5092, 'n_words': 1438, 'path': 'notebooks/scratch_chainable.ipynb', 'repo_name': 'hdmamin/htools', 'size': '16724'}