Invoke est un ordonnanceur et un exécuteur de tâches, c'est çà dire que il permet de décrire un workflow d'étapes pour atteindre un objectif et l'outil exécute celles-ci dans le bon ordre.

Il repose sur la définition des tâches (task) via l'annotation @task. Ces tâches sont définies au sein d'un fichier tasks.py.

tasks.py

from invoke import task
 
@task
def remove_all(ctx):
   ctx.run('rm -rf /tmp/my_dirs/work')

$ invoke -l
Available tasks:
 
  remove-all
 
$ invoke remove-all
$

Création de l'environnement virtuel de test

$ ~/Workdirs$ virtualenv -p python3 test_invoke
Running virtualenv with interpreter /usr/bin/python3
Using base prefix '/usr'
New python executable in /home/mdexet/Workdirs/test_invoke/bin/python3
Also creating executable in /home/mdexet/Workdirs/test_invoke/bin/python
Installing setuptools, pip, wheel...done.
$ ~/Workdirs$ cd test_invoke/
$ ~/Workdirs/test_invoke$ source ./bin/activate

Installation de pyInvoke

$ ~/Workdirs/test_invoke$ pip install invoke
Collecting invoke
  Downloading invoke-0.21.0-py3-none-any.whl (153kB)
    100% |████████████████████████████████| 153kB 3.9MB/s 
Installing collected packages: invoke
Successfully installed invoke-0.21.0

Pour créer une tâche il suffit

• de créer un fichier tasks.py
• de créer une fonction qui prend au moins un contexte ctx en argument

  • et de l'annoter avec @task

  <code python> @task def do_something(ctx): ….
  </code>

Une tâche peut exécuter une fonction python ou une commande système avec ctx.run(<commande>).

Appel de fonction

def do_it_right():
   //....
 
@task
def do_something(ctx):
   do_it_right()
 

Appel de commande système

@task(pre=[clean_workdir, get_last_commit, compile_project], 
      post=[deploy_artefact, update_website])
def build(ctx):
   ctx.run('maven install')

Un tâche peut nécessiter plusieurs taches en amont et/ou en aval du processus.

Pour modéliser cet enchaînement, il faut indiquer à la tache quels sont

• ses prédécesseurs avec la propriété pre

  • et ses successeurs avec la propriété post

  <code python> @task(pre=[cleanworkdir, getlastcommit, compileproject], post=[deployartefact, updatewebsite]) def build(ctx): do_something() </code>

La succession des tâches s'effectue de la première à la dernière sauf si

• une exception est levée
• une commande système ( ctx.run(<commande>) ) retourne un code erreur.

Exemple d'exception non gérée

@task
def get_last_commit(ctx):
    a = 10/0
    print('Get last commit')

$ invoke build 
Clean workdir
Get last commit
Traceback (most recent call last):
  File "/home/user/Workdirs/test_invoke/bin/invoke", line 11, in <module>
    sys.exit(program.run())
...
  File "/home/user/Workdirs/test_invoke/tasks.py", line 10, in get_last_commit
    a = 10/0
ZeroDivisionError: division by zero

Exemple d'exception gérée

@task
def get_last_commit(ctx):
    raise Exit('hostname not found')
    print('Get last commit')

$ invoke build 
Clean workdir
hostname not found

Pour désigner une tâche par défaut, il suffit de lui donner la propriété default=True

@task(default=True)
def build(ctx):
 ...

from invoke import task
 
@task
def clean_workdir(ctx):
    print('Clean workdir')
 
@task
def get_last_commit(ctx):
    print('Get last commit')
 
@task
def compile_project(ctx):
    print('Commit')
 
@task
def deploy_artefact(ctx):
    print('Deploy')
 
@task
def update_website(ctx):
    print('Update website')
 
@task(  pre=[clean_workdir, get_last_commit, compile_project],
        post=[deploy_artefact, update_website])
def build(ctx):
   print('Building')

$ invoke build
Clean workdir
Get last commit
Commit
Building
Deploy
Update website

Imaginons un projet avec

• un répertoire main/doc contenant la documentation en asciidoc
• un répertoire build/doc contenant la documentation finale
• un listing de fichiers dans main/src à mettre sous forme de document asciidoc

Décomposons le workflow.

Il faut

* générer un document //_listing.adoc// avec la liste des fichiers dans //main/src//
* générer un PDF dans //build/doc//
   * avant il faut s'assurer que //build/doc// existe

from invoke import task
import os
import shutil
from mako.template import Template
 
work_dirname = '/tmp/project_doc'
 
 
def collectfilenames(path):
    """
    Collect full qualified names of file within path.
    :param path: directory to scan
    :return: full qualified filenames from this path
    """
    path_list = []
    for dirpath, _, files_name in os.walk(path):
        for name in files_name:
            path_list.append("{}/{}".format(dirpath, name))
    return path_list
 
 
def recreate_dir(dirname):
    """
    Remove and recreate dirname
    :param dirname:
    """
    if os.path.exists(dirname):
        shutil.rmtree(dirname)
    os.makedirs(dirname)
 
 
@task
def move_in_tempdir(ctx):
    recreate_dir(work_dirname)
    with ctx.cd('main/doc'):
        ctx.run('cp * {}/'.format(work_dirname))
 
 
@task(pre=[move_in_tempdir])
def create_listing(ctx):
    with open('main/doc/_listing.adoc', 'w+') as listing:
        listing.write(Template(filename='main/doc/listing.tpl').render(filenames=collectfilenames('main/src')))
 
 
@task(move_in_tempdir, create_listing)
def generate_pdf(ctx):
    with ctx.cd(work_dirname):
        ctx.run('asciidoctor-pdf doc.adoc')
 
 
@task(pre=[generate_pdf], default=True)
def copy_to_builddir(ctx):
    dirname = 'build/doc'
 
    recreate_dir(dirname)
    with ctx.cd(dirname):
        ctx.run('cp {}/*.pdf  ./'.format(work_dirname))

Si une tâche doit retourner un résultat pour la tâche suivante, comment le passer ? Apparemment il n'est pas possible de transmettre un résultat par l'intermédiaire du contexte.

Le seul moyen est de peut-être utiliser un dictionnaire de stockage global.

storage = {}
 
@task
def do_something(ctx):
   storage['foo'] = 'bar'
 
 
@task(pre=[do_something])
def use_something(ctx):
    bar = storage['foo']

Exemple tiré de la documentation officielle

Supposons que nous ayons une commande qui demande des informations de façon interactive :

$ excitable-program
When you give the OK, I'm going to do the things. All of them!!
Are you ready? [Y/n] y
OK! I just did all sorts of neat stuff. You're welcome! Bye!

Pour cela il faut utiliser un Responder qui va surveiller la question et y répondre.

responder = Responder(pattern=r"Are you ready? \[Y/n\] ", response="y\n")
ctx.run("excitable-program", watchers=[responder])