@task doesn't play nice with other decorators

[lsh-0]

The @task decorator doesn't appear to play nicely with other decorators, unlike Fabric. I may not have read this part of the documentation yet, but I wouldn't expect @task to undermine python norms.

For example:

def capture_error(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except AssertionError as err:
            print(err)
    return wrapper

@task
@capture_error
def foo(c):
    print('bar')
    raise AssertionError('baz')

this looks fine:

inv foo
>>> bar
>>> baz

but when I add a parameter:

@task
@capture_error
def foo(c, p1):
    print(p1)
    raise AssertionError('baz')

I get:

inv foo bar
>>> No idea what 'bar' is!

The parameter handling docs offer no insight into what the problem is.

[bitprophet]

This is likely an oversight / lack of porting those extra careful parts of Fabric 1's decorator. #246 seems related and may fix it (though it's now kinda old so may need a revisit). Gonna leave this open and in a milestone since it's a regression (which is arguably worse than features which just aren't ported yet). Thanks!

[lsh-0]

hey, thanks for your attention on this. I've been using Fabric for so many years now I'm trialing Invoke on a personal project to unblock some work projects that need the Python3 upgrade as it doesn't look like Fabric will get there, which is frustrating, but I do appreciate your efforts and those of the other contributors.

[bitprophet]

@lsh-0 FWIW, Fabric 2 is technically out now (though it needs a few more feature releases to get up to par with 1.x) and supports Python 3. But it's also built heavily on top of Invoke so working with plain Invoke is a fine place to start; can always use fabric 2's API directly or switch later.

[ewanjones]

For future reference, I found a workaround:

def decorator(fn):
    @functools.wraps(fn)
    def _decorated(ctx, *args, **kwargs):
        return fn(ctx, *args, **kwargs)
    return _decorated

@invoke.task
def some_task(ctx, config="Something"):
    @decorator
    def _inner(ctx):
        print(config)
        ...

    return _inner(ctx)

This allows invoke.task to find the task and inspect the args

[corentinbettiol]

After finding your answer on stackoverflow I also found it here!

I was able to solve my problem by using the @functools.wraps decorator as well :

old decorator:

def ask(task_to_execute):
    """Decorator to ask if the user really wants to perform a task.
    """

    def wrapper(*args, **kwargs):

        y = ["y", "yes", "o", "oui", "Yes", "Oui", "go"]
        if input("Do you want to execute " + task_to_execute.__name__ + "? [y/d/N] ") in y:
            task_to_execute(*args, **kwargs)

    return wrapper

new decorator:

from functools import wraps

def ask(task_to_execute):
    """Decorator to ask if the user really wants to perform a task.
    """

    @wraps(task_to_execute)
    def wrapped_wrapper(*args, **kwargs):

        y = ["y", "yes", "o", "oui", "Yes", "Oui", "go"]
        if input("Do you want to execute " + task_to_execute.__name__ + "? [y/d/N] ") in y:
            task_to_execute(*args, **kwargs)

    return wrapped_wrapper

And so tab-completion works again :)

[iongion]

Any tip @corentinbettiol , I am getting 'No idea what 'basetask' is ?

@task
def basetask(ctx):
    print(ctx)

which I enhance with ask

@ask
@task
def basetask(ctx):
    print(ctx)

Then inv basetask => No idea what 'basetask' is!

[corentinbettiol]

Hi @iongion,

I'm using the two decorators in this order, that is different from your order:

@task()
@ask
def my_task(c, config):
    """Do something."""
    print("Do something")
    # ...

Maybe that's the problem?

---

That's completely unrelated, but now I display the docstring (it's Do something. here) by using task_to_execute.__doc__ in my ask function :)

[saveshodhan]

I had a similar issue, except that my task decorator also accepts arguments (for help), so the solution mentioned by @corentinbettiol didnt quite work for me. And i did not try @ewanjones's solution because i have a lot of functions and adding a decorator function inside each one will impact readability.

For now I came up with the below solution that works even when you have varags in your function and help kwarg in task decorator:

from functools import wraps
from inspect import signature

def handle_exit(task_to_execute):
    @wraps(task_to_execute)
    def inner(ctx, *a, **kw):
        try:
            return task_to_execute(ctx, *a, **kw)
        except UnexpectedExit:
            ctx.run(f"printf '\r\nMore info: {task_to_execute.__doc__}\r\n'", echo=False)
            raise
    inner.__signature__ = signature(task_to_execute)
    return inner

And then

@task(help={"host": "Host name", "i": "Number of packets to send"})
@handle_exit
def ping(ctx, host, i=1):
    """Check if host is pingable.

    If not, check with IT.
    """
    ctx.run(f"ping -c{i} -W1 -q {host}")

Output: Good case:

$ inv --echo ping --host www.google.com -i 3
ping -c3 -W1 -q www.google.com
PING www.google.com (142.250.180.68) 56(84) bytes of data.

--- www.google.com ping statistics ---
3 packets transmitted, 3 received, 0% packet loss, time 2004ms
rtt min/avg/max/mdev = 22.504/33.337/41.285/7.934 ms

Bad case:

$ inv --echo ping --host www.google.asdf -i 3
ping -c3 -W1 -q www.google.asdf
ping: www.google.asdf: Name or service not known

More info: Check if host is pingable.

    If not, check with IT.

Although I am still not sure why inner.__signature__ = signature(task_to_execute) worked, because if you add a pdb and see, signature(inner) and signature(task_to_execute) return the same values (<Signature (ctx, host, i=1)>) :man_shrugging:

[corentinbettiol]

@saveshodhan Ha! I also handled a "help text", but not like you!

If you look closely at my comment from 2021 you can spot a [y/d/N] ;

• The y is a str in my list (y = ["y", "yes", "o", "oui", "Yes", "Oui", "go"])
• The N (for no) is a str in another list (not displayed in my comment)
• And there's alors a d (for doc) that I forgot to remove from this (simplified) example!

Here's the real (complete) code I'm using:

def ask(task_to_execute):
    """Decorator to ask if the user really wants to perform a task.
    Possible answers: y (yes), d (doc, means "show me more doc about this"), and n (no).
    If command contain `--force`, execute function and don't ask anything.
    """

    @wraps(task_to_execute)
    def wrapped_wrapper(c, config, *args, **kwargs):
        if "--force" in argv:
            task_to_execute(c, config, *args, **kwargs)
        else:
            y = ["y", "yes", "o", "oui", "Yes", "Oui", "go"]
            d = ["d", "doc", "let me see the docstring for this function, please"]
            n = ["", "n", "no", "No", "non", "Non", "je refuse!"]
            answer = "not in n"

            while answer not in n:
                answer = input(
                    "Do you want to execute " + task_to_execute.__name__ + "? [y/d/N] "
                )

                if answer in d:
                    printunde(task_to_execute.__name__)
                    printdim(task_to_execute.__doc__)

                elif answer in y:
                    task_to_execute(c, config, *args, **kwargs)
                    break
    return wrapped_wrapper

I'm using tasks that call a lots of tasks in other namespaces (fab develop project-name will call the tasks git_clone/poetry_install_dependencies/gulp_compile_static/poetry_run_test_server, and all those tasks are in a namespace named local, so I can also do fab local.git_clone project-name), and sometimes I need to check the docstring to understand what a task is doing. So I came up with this [y/d/N] idea.

[saveshodhan]

Ah nice! Another problem i had was that different task functions have different function signatures - for eg., the ping function takes a host and count, but an ssh function might take different (and more) arguments. i tried with your solution and it worked when functions had fixed signatures, but for variable signatures i had to do the inner.__signature__ = signature(task_to_execute) thingy..

[corentinbettiol]

We're using fabric to execute a lot of tasks on our projects, and we choose to store all the project-specific informations in config files in each repo.

Each task takes the same nb of args, ans each task can look into the config var that holds our project config :)

So here's the real real (complete) code we're using :

from functools import wraps
from os import path
from sys import argv

import toml

from .utils.colors import printdim, printfail, printokb, printunde

def read_config_project(c, name):
    if path.isfile(name + "/config.toml"):
        config = toml.load(name + "/config.toml")
        return config
    else:
        printfail("No config file found!")
    exit(1)

def ask(task_to_execute):
    """Decorator to ask if the user really wants to perform a task.
    Possible answers: y (yes), d (doc, means "show me more doc about this"), and n (no).
    If command contain `--force`, execute function and don't ask anything.
    """

    @wraps(task_to_execute)
    def wrapped_wrapper(c, config, *args, **kwargs):
        # If config is a str (the name of a project) = if subtask was launched by command line
        # directly instead of a "mega task".
        # In this case the config.toml fils is not loaded, so we need to do it here.
        # We need the name of the project to search the config file in the project folder.
        if type(config) == str:
            print("Project ", end="")
            printunde(config)
            printdim("Config not found. Loading config.toml... ", end="")
            config = read_config_project(c, config)
            print("ok!")

        if "--force" in argv:
            task_to_execute(c, config, *args, **kwargs)
        else:
            y = ["y", "yes", "o", "oui", "Yes", "Oui", "go"]
            d = ["d", "doc", "let me see the docstring for this function, please"]
            n = ["", "n", "no", "No", "non", "Non", "je refuse!"]
            answer = "not in n"

            while answer not in n:
                answer = input(
                    "Do you want to execute " + task_to_execute.__name__ + "? [y/d/N] "
                )

                if answer in d:
                    printunde(task_to_execute.__name__)
                    printdim(task_to_execute.__doc__)

                elif answer in y:
                    task_to_execute(c, config, *args, **kwargs)
                    break
    return wrapped_wrapper