Python invoke: Add --exec option and make tasks dryable.

example_tasks.py

import invoke
import invoke_dryable as dryable


@invoke.task
@dryable.default_dryrun
def build(c, clean=False):
    if clean:
        print("Cleaning!")

    print("Building!")

    c.run('echo hoge')

invoke_dryable.py

__all__ = [
    'default_dryrun',
    'default_exec',
]


import functools
import inspect
import invoke
import itertools


###
### util
###

## From doc of itertools

def find(iterable, default=None, pred=None):
    '''
    Returns the first true value in the iterable.
    If no true value is found, returns *default*
    If *pred* is not None, returns the first item
    for which pred(item) is true.
    '''
    # find([a,b,c], x) --> a or b or c or x
    # find([a,b], x, f) --> a if f(a) else b if f(b) else x
    return next(filter(pred, iterable), default)


## From SRFI-1

def list_index(iterable, default=None, pred=None):
    '''
    Return left-most index in list satisfying `pred`.
    This is a lambda version of `List.index`.
    '''
    # list_index(['a', 'b', 'c'], None, lambda s: s == 'b')
    # => 1
    # list_index(['a', 'b', 'c'], 100, lambda s: s == 'x')
    # => 100
    return find(zip(itertools.count(start=0, step=1), iterable), default=(default,), pred=lambda x: pred(x[1]))[0]


###
### main
###


# class DryableContext(invoke.Context):
class DryableContext:
    def __init__(self, context, is_dryrun):
        self.__dict__ = {
            '_context': context,
            '_is_dryrun': is_dryrun,
        }

    def __getattr__(self, key):
        if self.__dict__['_is_dryrun'] and (key in ['run']):
            return self._dryrun
        else:
            return getattr(self._context, key)

    def _dryrun(self, *args, **keywords):
        print(f"DRYRUN: {args}")


def default_dryrun(def_task):
    @functools.wraps(def_task)
    def modified(c, *args, exec=False, **keywords):
        c = DryableContext(c, not exec)
        def_task(c, *args, **keywords)

    added_param = inspect.Parameter(
        'exec',
        inspect.Parameter.POSITIONAL_OR_KEYWORD,
        default=False
    )
    orig_params = list(inspect.signature(def_task).parameters.values())
    KEYWORD_ONLY = inspect.Parameter.KEYWORD_ONLY
    VAR_KEYWORD = inspect.Parameter.VAR_KEYWORD
    i = list_index(orig_params, default=len(orig_params), pred=lambda p: (p.kind == KEYWORD_ONLY) or (p.kind == VAR_KEYWORD))
    params = orig_params[:i] + [added_param] + orig_params[i:]
    modified.__signature__ = inspect.Signature(params)

    return modified


def default_exec(def_task):
    @functools.wraps(def_task)
    def modified(c, *args, dryrun=False, **keywords):
        c = DryableContext(c, dryrun)
        def_task(c, *args, **keywords)

    added_param = inspect.Parameter(
        'dryrun',
        inspect.Parameter.POSITIONAL_OR_KEYWORD,
        default=False
    )
    orig_params = list(inspect.signature(def_task).parameters.values())
    KEYWORD_ONLY = inspect.Parameter.KEYWORD_ONLY
    VAR_KEYWORD = inspect.Parameter.VAR_KEYWORD
    i = list_index(orig_params, default=len(orig_params), pred=lambda p: (p.kind == KEYWORD_ONLY) or (p.kind == VAR_KEYWORD))
    params = orig_params[:i] + [added_param] + orig_params[i:]
    modified.__signature__ = inspect.Signature(params)

    return modified